amadan

carlos / platform Public

applog: the pure half of per-app log shipping

Record (additive wire shape), ParseAppLine (structured lines kept, levels
never guessed), LineWriter (the exec tee's line splitter), Sink (ring +
rate budgets with drop honesty), gzipped-JSONL chunk codec + key builders
under the app prefix, and the read-side Filter/Merge fold. Spec
2026-08-10-carlos-logs-design.md §§3-5.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell b16440ce6734042359875cba259fac8e4c6ffda5 parent 6bd3b7a
8 files changed, +1050 −0
  • internal/applog/chunk.go +94 −0
  • internal/applog/chunk_test.go +110 −0
  • internal/applog/filter.go +108 −0
  • internal/applog/filter_test.go +91 −0
  • internal/applog/record.go +215 −0
  • internal/applog/record_test.go +133 −0
  • internal/applog/sink.go +163 −0
  • internal/applog/sink_test.go +136 −0
diff --git a/internal/applog/chunk.go b/internal/applog/chunk.go
new file mode 100644
index 000000000..ad53ffffe
--- /dev/null
+++ b/internal/applog/chunk.go
@@ -0,0 +1,94 @@
+package applog
+
+import (
+ "bufio"
+ "bytes"
+ "compress/gzip"
+ "encoding/json"
+ "fmt"
+ "io"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// hourFormat is the UTC hour vocabulary in chunk keys — deliberately the
+// same format internal/metrics uses for its raw chunks, so the bucket
+// speaks one hour dialect.
+const hourFormat = "2006-01-02-15"
+
+// HourOf formats t as the UTC hour directory a chunk belongs to.
+func HourOf(t time.Time) string { return t.UTC().Format(hourFormat) }
+
+// HourBefore reports whether hour (a HourOf-formatted directory name) ends
+// strictly before cutoff's own hour — the retention test. An unparsable
+// name never matches: pruning must not delete on a guess.
+func HourBefore(hour string, cutoff time.Time) bool {
+ t, err := time.ParseInLocation(hourFormat, hour, time.UTC)
+ if err != nil {
+ return false
+ }
+ return t.Add(time.Hour).Sub(cutoff.UTC()) <= 0 && HourOf(cutoff) != hour
+}
+
+// Prefix is the app's whole log subtree — under release.AppPrefix on
+// purpose (spec §3): logs are the app's data, so the owner gate maps onto
+// the prefix and app trash/purge sweeps them for free.
+func Prefix(acct, app string) string { return release.AppPrefix(acct, app) + "logs/" }
+
+// BoxPrefix is one box's subtree within an app's logs — the unit of
+// write-side ownership: each box prunes only what it wrote.
+func BoxPrefix(acct, app, box string) string { return Prefix(acct, app) + box + "/" }
+
+// HourPrefix is one UTC hour within a box's subtree.
+func HourPrefix(acct, app, box, hour string) string {
+ return BoxPrefix(acct, app, box) + hour + "/"
+}
+
+// ChunkKey names one flushed chunk: seq is the first record's per-box
+// sequence number, zero-padded so lexical listing order is numeric order.
+// Chunks are written once and never appended or rewritten (spec §3).
+func ChunkKey(acct, app, box, hour string, seq uint64) string {
+ return fmt.Sprintf("%s%012d.jsonl.gz", HourPrefix(acct, app, box, hour), seq)
+}
+
+// EncodeChunk renders records as gzipped JSONL, one record per line.
+func EncodeChunk(recs []Record) ([]byte, error) {
+ var buf bytes.Buffer
+ zw := gzip.NewWriter(&buf)
+ enc := json.NewEncoder(zw)
+ for _, r := range recs {
+ if err := enc.Encode(r); err != nil {
+ return nil, err
+ }
+ }
+ if err := zw.Close(); err != nil {
+ return nil, err
+ }
+ return buf.Bytes(), nil
+}
+
+// DecodeChunk parses a chunk, skipping malformed lines rather than failing
+// the whole object — one corrupt line must not cost the member the rest of
+// the hour. Corrupt gzip is an error: that is a broken object, not a line.
+func DecodeChunk(b []byte) ([]Record, error) {
+ zr, err := gzip.NewReader(bytes.NewReader(b))
+ if err != nil {
+ return nil, err
+ }
+ defer zr.Close()
+ var out []Record
+ sc := bufio.NewScanner(zr)
+ sc.Buffer(nil, 1<<20) // lines are capped well below this at the tee
+ for sc.Scan() {
+ var r Record
+ if json.Unmarshal(sc.Bytes(), &r) != nil {
+ continue
+ }
+ out = append(out, r)
+ }
+ if err := sc.Err(); err != nil && err != io.EOF {
+ return out, err
+ }
+ return out, nil
+}
diff --git a/internal/applog/chunk_test.go b/internal/applog/chunk_test.go
new file mode 100644
index 000000000..137b6a6f5
--- /dev/null
+++ b/internal/applog/chunk_test.go
@@ -0,0 +1,110 @@
+package applog
+
+import (
+ "bytes"
+ "compress/gzip"
+ "io"
+ "testing"
+ "time"
+)
+
+func mustGunzip(t *testing.T, b []byte) []byte {
+ t.Helper()
+ zr, err := gzip.NewReader(bytes.NewReader(b))
+ if err != nil {
+ t.Fatal(err)
+ }
+ out, err := io.ReadAll(zr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return out
+}
+
+func mustGzip(t *testing.T, b []byte) []byte {
+ t.Helper()
+ var buf bytes.Buffer
+ zw := gzip.NewWriter(&buf)
+ if _, err := zw.Write(b); err != nil {
+ t.Fatal(err)
+ }
+ if err := zw.Close(); err != nil {
+ t.Fatal(err)
+ }
+ return buf.Bytes()
+}
+
+func TestKeysAndHourFormat(t *testing.T) {
+ at := time.Date(2026, 8, 10, 21, 59, 3, 0, time.UTC)
+ if got := HourOf(at); got != "2026-08-10-21" {
+ t.Fatalf("HourOf = %q", got) // same vocabulary as metrics' hourFormat
+ }
+ if got := Prefix("bde", "lpay"); got != "apps/bde/lpay/logs/" {
+ t.Fatalf("Prefix = %q", got)
+ }
+ if got := BoxPrefix("bde", "lpay", "box1"); got != "apps/bde/lpay/logs/box1/" {
+ t.Fatalf("BoxPrefix = %q", got)
+ }
+ if got := HourPrefix("bde", "lpay", "box1", "2026-08-10-21"); got != "apps/bde/lpay/logs/box1/2026-08-10-21/" {
+ t.Fatalf("HourPrefix = %q", got)
+ }
+ // Zero-padded so lexical order is numeric order.
+ if got := ChunkKey("bde", "lpay", "box1", "2026-08-10-21", 41822); got != "apps/bde/lpay/logs/box1/2026-08-10-21/000000041822.jsonl.gz" {
+ t.Fatalf("ChunkKey = %q", got)
+ }
+}
+
+func TestChunkRoundTrip(t *testing.T) {
+ in := []Record{
+ {TS: time.Date(2026, 8, 10, 21, 0, 1, 0, time.UTC), Box: "b1", Host: "h", Stream: StreamApp, Msg: "one", Seq: 1},
+ {TS: time.Date(2026, 8, 10, 21, 0, 2, 0, time.UTC), Box: "b1", Stream: StreamPlatform, Level: LevelError, Msg: "two", Seq: 2, Fields: map[string]string{"k": "v"}},
+ }
+ b, err := EncodeChunk(in)
+ if err != nil {
+ t.Fatal(err)
+ }
+ out, err := DecodeChunk(b)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(out) != 2 || !out[0].TS.Equal(in[0].TS) || out[1].Fields["k"] != "v" || out[1].Seq != 2 {
+ t.Fatalf("round trip lost data: %+v", out)
+ }
+}
+
+// TestDecodeChunkSkipsMalformedLines: one corrupt line must not cost the
+// member the rest of the hour — decode keeps what parses.
+func TestDecodeChunkSkipsMalformedLines(t *testing.T) {
+ good := []Record{{TS: time.Date(2026, 8, 10, 21, 0, 1, 0, time.UTC), Msg: "ok", Seq: 9}}
+ b, err := EncodeChunk(good)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Splice garbage into the gzip payload by re-encoding by hand.
+ raw := append([]byte("this is not json\n"), mustGunzip(t, b)...)
+ out, err := DecodeChunk(mustGzip(t, raw))
+ if err != nil {
+ t.Fatalf("decode failed outright on one bad line: %v", err)
+ }
+ if len(out) != 1 || out[0].Msg != "ok" {
+ t.Fatalf("good line lost: %+v", out)
+ }
+ // Truly corrupt gzip IS an error — that's a broken object, not a line.
+ if _, err := DecodeChunk([]byte("not gzip at all")); err == nil {
+ t.Fatal("corrupt chunk decoded to no error")
+ }
+}
+
+func TestHourBefore(t *testing.T) {
+ cutoff := time.Date(2026, 8, 10, 21, 30, 0, 0, time.UTC)
+ for hour, want := range map[string]bool{
+ "2026-08-10-20": true, // strictly before the cutoff's hour
+ "2026-08-10-21": false, // the cutoff's own hour is live
+ "2026-08-10-22": false,
+ "garbage": false, // unparsable never matches — never delete on a guess
+ } {
+ if got := HourBefore(hour, cutoff); got != want {
+ t.Errorf("HourBefore(%q) = %v, want %v", hour, got, want)
+ }
+ }
+}
diff --git a/internal/applog/filter.go b/internal/applog/filter.go
new file mode 100644
index 000000000..a6eb235e2
--- /dev/null
+++ b/internal/applog/filter.go
@@ -0,0 +1,108 @@
+package applog
+
+import (
+ "fmt"
+ "regexp"
+ "sort"
+ "strings"
+ "time"
+)
+
+// Filter is the read-side match: zero values mean "no constraint", so the
+// zero Filter matches everything. Applied console-side (spec §6) so the
+// whole window never crosses the wire just to be discarded.
+type Filter struct {
+ Since time.Time // inclusive; zero = unbounded
+ Until time.Time // inclusive; zero = unbounded
+ Stream string
+ Host string
+ // Level is a threshold, not an exact match: "warn" admits warn and
+ // error. A record with no level NEVER satisfies a set threshold —
+ // levels are never guessed (spec §4), so an unleveled record cannot
+ // claim to clear a bar it never set.
+ Level string
+ Grep *regexp.Regexp // matched against Msg
+ Fields map[string]string // every listed key must be present and equal
+}
+
+// Match reports whether r survives f.
+func (f Filter) Match(r Record) bool {
+ if !f.Since.IsZero() && r.TS.Before(f.Since) {
+ return false
+ }
+ if !f.Until.IsZero() && r.TS.After(f.Until) {
+ return false
+ }
+ if f.Stream != "" && r.Stream != f.Stream {
+ return false
+ }
+ if f.Host != "" && r.Host != f.Host {
+ return false
+ }
+ if f.Level != "" {
+ want := levelRank(f.Level)
+ got := levelRank(r.Level)
+ if got < 0 || got < want {
+ return false
+ }
+ }
+ if f.Grep != nil && !f.Grep.MatchString(r.Msg) {
+ return false
+ }
+ for k, v := range f.Fields {
+ if r.Fields[k] != v {
+ return false
+ }
+ }
+ return true
+}
+
+// MergeRecords folds per-source slices into one timeline ordered by
+// (TS, Box, Seq). Per-box order is exact (Seq is monotonic per box);
+// cross-box order is best-effort by box clock with a deterministic
+// tiebreak — the only promise the spec makes (§4).
+func MergeRecords(slices ...[]Record) []Record {
+ var out []Record
+ for _, s := range slices {
+ out = append(out, s...)
+ }
+ sort.SliceStable(out, func(i, j int) bool {
+ a, b := out[i], out[j]
+ if !a.TS.Equal(b.TS) {
+ return a.TS.Before(b.TS)
+ }
+ if a.Box != b.Box {
+ return a.Box < b.Box
+ }
+ return a.Seq < b.Seq
+ })
+ return out
+}
+
+// ParseTimeArg reads a -since/-until value: RFC3339, or a Go duration
+// shorthand ("15m", "1h") meaning that long before now. Empty is the zero
+// time (no constraint), not an error.
+func ParseTimeArg(s string, now time.Time) (time.Time, error) {
+ if s == "" {
+ return time.Time{}, nil
+ }
+ if d, err := time.ParseDuration(s); err == nil {
+ if d < 0 {
+ return time.Time{}, fmt.Errorf("negative duration %q", s)
+ }
+ return now.Add(-d), nil
+ }
+ if t, err := time.Parse(time.RFC3339, s); err == nil {
+ return t, nil
+ }
+ return time.Time{}, fmt.Errorf("cannot parse %q: want RFC3339 (2026-08-10T11:00:00Z) or a duration (15m, 1h)", s)
+}
+
+// ParseFieldArg reads one -field k=v argument.
+func ParseFieldArg(s string) (k, v string, err error) {
+ k, v, ok := strings.Cut(s, "=")
+ if !ok || k == "" {
+ return "", "", fmt.Errorf("cannot parse field %q: want key=value", s)
+ }
+ return k, v, nil
+}
diff --git a/internal/applog/filter_test.go b/internal/applog/filter_test.go
new file mode 100644
index 000000000..ac08b3d04
--- /dev/null
+++ b/internal/applog/filter_test.go
@@ -0,0 +1,91 @@
+package applog
+
+import (
+ "regexp"
+ "testing"
+ "time"
+)
+
+func at(sec int) time.Time { return time.Date(2026, 8, 10, 12, 0, sec, 0, time.UTC) }
+
+func TestFilterMatch(t *testing.T) {
+ rec := Record{
+ TS: at(30), Box: "b1", Host: "lpay.bde.oncarlos.com",
+ Stream: StreamApp, Level: LevelWarn, Msg: "slow query on /checkout",
+ Fields: map[string]string{"request_id": "r1"},
+ }
+ cases := []struct {
+ name string
+ f Filter
+ want bool
+ }{
+ {"zero filter matches", Filter{}, true},
+ {"since inclusive window", Filter{Since: at(30)}, true},
+ {"since excludes older", Filter{Since: at(31)}, false},
+ {"until excludes newer", Filter{Until: at(29)}, false},
+ {"stream match", Filter{Stream: StreamApp}, true},
+ {"stream mismatch", Filter{Stream: StreamPlatform}, false},
+ {"host match", Filter{Host: "lpay.bde.oncarlos.com"}, true},
+ {"host mismatch", Filter{Host: "other"}, false},
+ {"level threshold met", Filter{Level: LevelWarn}, true},
+ {"level below threshold", Filter{Level: LevelError}, false},
+ {"grep on msg", Filter{Grep: regexp.MustCompile(`checkout`)}, true},
+ {"grep miss", Filter{Grep: regexp.MustCompile(`nope`)}, false},
+ {"field match", Filter{Fields: map[string]string{"request_id": "r1"}}, true},
+ {"field mismatch", Filter{Fields: map[string]string{"request_id": "r2"}}, false},
+ {"field absent", Filter{Fields: map[string]string{"user": "u"}}, false},
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ if got := tc.f.Match(rec); got != tc.want {
+ t.Fatalf("Match = %v, want %v", got, tc.want)
+ }
+ })
+ }
+ // A record with NO level never satisfies a level threshold — levels are
+ // never guessed, so it cannot claim to clear a bar it never set.
+ unleveled := rec
+ unleveled.Level = ""
+ if (Filter{Level: LevelDebug}).Match(unleveled) {
+ t.Fatal("unleveled record passed a level threshold")
+ }
+}
+
+// TestMergeRecords: cross-box merge orders by (ts, box, seq) — per-box
+// order preserved, best-effort across boxes, deterministic tiebreak.
+func TestMergeRecords(t *testing.T) {
+ b1 := []Record{
+ {TS: at(1), Box: "b1", Seq: 10, Msg: "b1-first"},
+ {TS: at(3), Box: "b1", Seq: 11, Msg: "b1-second"},
+ }
+ b2 := []Record{
+ {TS: at(2), Box: "b2", Seq: 5, Msg: "b2-first"},
+ {TS: at(3), Box: "a0", Seq: 1, Msg: "a0-tiebreak"}, // same ts as b1-second
+ }
+ got := MergeRecords(b1, b2)
+ want := []string{"b1-first", "b2-first", "a0-tiebreak", "b1-second"}
+ if len(got) != len(want) {
+ t.Fatalf("merged %d records", len(got))
+ }
+ for i, m := range want {
+ if got[i].Msg != m {
+ t.Fatalf("order[%d] = %q, want %q (full: %+v)", i, got[i].Msg, m, got)
+ }
+ }
+}
+
+func TestParseSinceUntil(t *testing.T) {
+ now := time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)
+ if got, err := ParseTimeArg("15m", now); err != nil || !got.Equal(now.Add(-15*time.Minute)) {
+ t.Fatalf("duration shorthand: %v %v", got, err)
+ }
+ if got, err := ParseTimeArg("2026-08-10T11:00:00Z", now); err != nil || got.Hour() != 11 {
+ t.Fatalf("RFC3339: %v %v", got, err)
+ }
+ if _, err := ParseTimeArg("yesterday-ish", now); err == nil {
+ t.Fatal("garbage parsed")
+ }
+ if got, err := ParseTimeArg("", now); err != nil || !got.IsZero() {
+ t.Fatalf("empty should be zero time, no error: %v %v", got, err)
+ }
+}
diff --git a/internal/applog/record.go b/internal/applog/record.go
new file mode 100644
index 000000000..8e08f14c6
--- /dev/null
+++ b/internal/applog/record.go
@@ -0,0 +1,215 @@
+// Package applog is the pure half of per-app log shipping (`carlos logs`,
+// spec 2026-08-10-carlos-logs-design.md): the wire record, the box-side
+// sink (ring + budgets + drop honesty), the gzipped-JSONL chunk codec and
+// bucket key builders, and the read-side filter/merge fold. No I/O and no
+// clock of its own — internal/host owns the shipping loop, internal/console
+// owns the read path, exactly the internal/metrics split.
+package applog
+
+import (
+ "encoding/json"
+ "strconv"
+ "strings"
+ "time"
+)
+
+// Streams a Record can belong to. The vocabulary is fixed by the spec (§1)
+// so later stages are additive: stage 1 ships app + platform; edge and
+// release exist in the vocabulary now so nothing re-parses when they arrive.
+const (
+ StreamApp = "app" // the app process's own stdout/stderr
+ StreamPlatform = "platform" // wake/restart/lifecycle events, member-safe text
+ StreamEdge = "edge" // per-request access lines (stage 2)
+ StreamRelease = "release" // ship/promote history join (stage 3)
+)
+
+// ValidStream reports whether s names a known stream — the CLI/API's
+// vocabulary check, kept here so there is exactly one list.
+func ValidStream(s string) bool {
+ switch s {
+ case StreamApp, StreamPlatform, StreamEdge, StreamRelease:
+ return true
+ }
+ return false
+}
+
+// Levels a Record may carry. Order matters to Filter's threshold semantics;
+// levelRank is the one place that order lives.
+const (
+ LevelDebug = "debug"
+ LevelInfo = "info"
+ LevelWarn = "warn"
+ LevelError = "error"
+)
+
+// levelRank maps a level to its severity order. Unknown/absent levels rank
+// -1: a record that never claimed a level can never satisfy a -level
+// threshold, because a guessed level is a confident wrong answer (spec §4).
+func levelRank(level string) int {
+ switch level {
+ case LevelDebug:
+ return 0
+ case LevelInfo:
+ return 1
+ case LevelWarn:
+ return 2
+ case LevelError:
+ return 3
+ }
+ return -1
+}
+
+// ValidLevel reports whether s names a known level.
+func ValidLevel(s string) bool { return levelRank(s) >= 0 }
+
+// Record is one log line on the wire — one JSONL line inside a chunk.
+// Additive shape (CLAUDE.md rule): every field but TS and Msg is omitempty,
+// unknown keys are ignored on read, and anything written before a shape
+// change must still parse. TS is the box clock in UTC; cross-box order is
+// best-effort by TS with (Box, Seq) as tiebreak — per-box order is the only
+// promise. Seq is monotonic per box process, which is what makes the follow
+// cursor exact and gaps detectable rather than invisible.
+type Record struct {
+ TS time.Time `json:"ts"`
+ Box string `json:"box,omitempty"`
+ Host string `json:"host,omitempty"` // route host — how owners name instances
+ Stream string `json:"stream,omitempty"`
+ Level string `json:"level,omitempty"`
+ Msg string `json:"msg"`
+ Fields map[string]string `json:"fields,omitempty"`
+ Seq uint64 `json:"seq,omitempty"`
+}
+
+// ParseAppLine interprets one line of app output. A line that parses as a
+// top-level JSON object yields: fields from its scalar values (numbers and
+// bools stringified; nested values ignored rather than flattened), msg from
+// a string-valued "msg" or "message" key (the raw line when neither is
+// present), and level ONLY when a "level" key carries a recognized value
+// (case-insensitive; "warning" counts as warn) — an unrecognized level
+// stays in fields as data. Anything else is opaque text: msg is the line,
+// untouched. Level is never guessed from text (spec §4): a guessed level
+// that makes -level error drop the crash line is worse than no level.
+func ParseAppLine(line string) (msg, level string, fields map[string]string) {
+ trimmed := strings.TrimSpace(line)
+ if !strings.HasPrefix(trimmed, "{") {
+ return line, "", nil
+ }
+ var obj map[string]any
+ if err := json.Unmarshal([]byte(trimmed), &obj); err != nil {
+ return line, "", nil
+ }
+ msg = line
+ for _, key := range []string{"msg", "message"} {
+ if s, ok := obj[key].(string); ok {
+ msg = s
+ delete(obj, key)
+ break
+ }
+ }
+ if s, ok := obj["level"].(string); ok {
+ if l := normalizeLevel(s); l != "" {
+ level = l
+ delete(obj, "level")
+ }
+ }
+ for k, v := range obj {
+ var s string
+ switch v := v.(type) {
+ case string:
+ s = v
+ case bool:
+ s = strconv.FormatBool(v)
+ case float64:
+ s = strconv.FormatFloat(v, 'f', -1, 64)
+ default: // nested object/array/null — not flattened, not invented
+ continue
+ }
+ if fields == nil {
+ fields = make(map[string]string)
+ }
+ fields[k] = s
+ }
+ return msg, level, fields
+}
+
+// normalizeLevel folds the common spellings onto the fixed vocabulary,
+// answering "" for anything it does not recognize.
+func normalizeLevel(s string) string {
+ switch strings.ToLower(s) {
+ case "debug", "trace":
+ return LevelDebug
+ case "info":
+ return LevelInfo
+ case "warn", "warning":
+ return LevelWarn
+ case "error", "err", "fatal", "panic":
+ return LevelError
+ }
+ return ""
+}
+
+// truncMarker is appended to a line NewLineWriter cut at its cap. It reads
+// as what happened rather than pretending the line ended there.
+const truncMarker = "…(truncated)"
+
+// LineWriter adapts a byte stream (an exec child's stdout/stderr) into
+// whole log lines: split on \n, blank lines dropped, over-long lines cut at
+// max bytes with truncMarker appended and the remainder of that line
+// discarded (not emitted as a phantom second line). Close flushes a
+// trailing partial so a crash's last unterminated words still ship.
+//
+// Not safe for concurrent Write from multiple goroutines — each stream gets
+// its own LineWriter, which is also what keeps interleaving per-stream.
+type LineWriter struct {
+ max int
+ emit func(line string)
+ buf []byte
+ skipping bool // inside the discarded tail of an over-long line
+}
+
+// NewLineWriter returns a LineWriter emitting complete lines to emit, each
+// at most max bytes before the truncation marker.
+func NewLineWriter(max int, emit func(line string)) *LineWriter {
+ return &LineWriter{max: max, emit: emit}
+}
+
+// Write is infallible by design: a log tee must never be the reason an
+// app's stdout write fails.
+func (w *LineWriter) Write(p []byte) (int, error) {
+ for _, b := range p {
+ if b == '\n' {
+ w.flush()
+ continue
+ }
+ if w.skipping {
+ continue
+ }
+ w.buf = append(w.buf, b)
+ if len(w.buf) >= w.max {
+ w.emit(string(w.buf) + truncMarker)
+ w.buf = w.buf[:0]
+ w.skipping = true
+ }
+ }
+ return len(p), nil
+}
+
+// Close flushes any trailing partial line. It never errors; the signature
+// exists so a LineWriter can stand where an io.Closer is expected.
+func (w *LineWriter) Close() error {
+ w.flush()
+ return nil
+}
+
+func (w *LineWriter) flush() {
+ if w.skipping {
+ w.skipping = false
+ w.buf = w.buf[:0]
+ return
+ }
+ if len(w.buf) == 0 {
+ return
+ }
+ w.emit(string(w.buf))
+ w.buf = w.buf[:0]
+}
diff --git a/internal/applog/record_test.go b/internal/applog/record_test.go
new file mode 100644
index 000000000..ee89c54ae
--- /dev/null
+++ b/internal/applog/record_test.go
@@ -0,0 +1,133 @@
+package applog
+
+import (
+ "encoding/json"
+ "strings"
+ "testing"
+ "time"
+)
+
+// TestRecordWireShapeIsAdditive pins the wire contract: every field but ts
+// and msg is omitempty, and a record written by a future revision with keys
+// this one has never heard of still parses. Live chunks exist in production
+// buckets from the moment this ships — anything written before a change
+// must still parse (CLAUDE.md wire-shape rule).
+func TestRecordWireShapeIsAdditive(t *testing.T) {
+ r := Record{
+ TS: time.Date(2026, 8, 10, 14, 3, 11, 482913000, time.UTC),
+ Msg: "hello",
+ }
+ b, err := json.Marshal(r)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, absent := range []string{"box", "host", "stream", "level", "fields", "seq"} {
+ if strings.Contains(string(b), `"`+absent+`"`) {
+ t.Errorf("zero-valued %q should be omitted from the wire: %s", absent, b)
+ }
+ }
+
+ // A future record with unknown keys parses; known keys land.
+ future := `{"ts":"2026-08-10T14:03:11.482913Z","msg":"m","stream":"app",` +
+ `"seq":41822,"box":"b1","host":"h","level":"error",` +
+ `"fields":{"k":"v"},"shiny_new_thing":{"deep":true}}`
+ var got Record
+ if err := json.Unmarshal([]byte(future), &got); err != nil {
+ t.Fatalf("future record must parse: %v", err)
+ }
+ if got.Seq != 41822 || got.Stream != StreamApp || got.Fields["k"] != "v" {
+ t.Fatalf("known fields lost in additive parse: %+v", got)
+ }
+}
+
+// TestParseAppLine covers the JSON-line contract: a top-level JSON object
+// yields fields from its scalar values, msg from msg/message, level only
+// from a recognized value — and anything else is opaque text, untouched.
+func TestParseAppLine(t *testing.T) {
+ cases := []struct {
+ name, line string
+ wantMsg, wantLevel string
+ wantFields map[string]string
+ }{
+ {"plain text", "listening on :8080", "listening on :8080", "", nil},
+ {"non-object json", `[1,2,3]`, `[1,2,3]`, "", nil},
+ {
+ "structured",
+ `{"level":"ERROR","msg":"boom","request_id":"a1b2","n":7,"ok":true}`,
+ "boom", "error",
+ map[string]string{"request_id": "a1b2", "n": "7", "ok": "true"},
+ },
+ {
+ "message key and unknown level",
+ `{"level":"shouting","message":"hi"}`,
+ "hi", "",
+ // An unrecognized level is DATA, not a level — it must survive
+ // as a field rather than be guessed at or dropped.
+ map[string]string{"level": "shouting"},
+ },
+ {
+ "nested values are not flattened",
+ `{"msg":"m","deep":{"a":1}}`,
+ "m", "", nil,
+ },
+ {"empty line", "", "", "", nil},
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ msg, level, fields := ParseAppLine(tc.line)
+ if msg != tc.wantMsg || level != tc.wantLevel {
+ t.Fatalf("msg=%q level=%q, want %q/%q", msg, level, tc.wantMsg, tc.wantLevel)
+ }
+ if len(fields) != len(tc.wantFields) {
+ t.Fatalf("fields=%v, want %v", fields, tc.wantFields)
+ }
+ for k, v := range tc.wantFields {
+ if fields[k] != v {
+ t.Fatalf("fields[%q]=%q, want %q", k, fields[k], v)
+ }
+ }
+ })
+ }
+}
+
+// TestLineWriter: the tee's writer half splits on \n, caps line length with
+// an explicit truncation marker, and flushes a trailing partial on Close —
+// output bytes are never reordered or dropped silently.
+func TestLineWriter(t *testing.T) {
+ var got []string
+ w := NewLineWriter(16, func(line string) { got = append(got, line) })
+
+ // Split across writes; only complete lines emit.
+ w.Write([]byte("hel"))
+ w.Write([]byte("lo\nwor"))
+ if len(got) != 1 || got[0] != "hello" {
+ t.Fatalf("after partial writes got %q", got)
+ }
+ // An over-long line is truncated once, marked, and the tail discarded —
+ // not emitted as a phantom second line.
+ w.Write([]byte("ldxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx-tail\nnext\n"))
+ if len(got) != 3 {
+ t.Fatalf("got %q", got)
+ }
+ if !strings.HasSuffix(got[1], "…(truncated)") || len(got[1]) > 16+len("…(truncated)") {
+ t.Fatalf("long line not truncated with marker: %q", got[1])
+ }
+ if got[2] != "next" {
+ t.Fatalf("line after a truncated one corrupted: %q", got[2])
+ }
+ // Close flushes the partial.
+ w.Write([]byte("no newline"))
+ w.Close()
+ if got[len(got)-1] != "no newline" {
+ t.Fatalf("Close did not flush the partial: %q", got)
+ }
+ // Empty lines are dropped (a trailing \n on every Printf would
+ // otherwise ship a blank record each time).
+ got = nil
+ w2 := NewLineWriter(16, func(line string) { got = append(got, line) })
+ w2.Write([]byte("\n\n"))
+ w2.Close()
+ if len(got) != 0 {
+ t.Fatalf("blank lines shipped: %q", got)
+ }
+}
diff --git a/internal/applog/sink.go b/internal/applog/sink.go
new file mode 100644
index 000000000..0f7490796
--- /dev/null
+++ b/internal/applog/sink.go
@@ -0,0 +1,163 @@
+package applog
+
+import (
+ "fmt"
+ "sync"
+ "time"
+)
+
+// AppKey names the (account, app) a buffered record belongs to — the unit
+// both budgets and bucket keys are scoped by.
+type AppKey struct{ Acct, App string }
+
+// Sink buffers tagged records between ship-loop flushes: a bounded ring per
+// app plus a token-bucket rate cap per app, both with drop honesty — when
+// either binds, oldest lines go first and the next Drain carries one
+// synthetic platform record naming the count, never silent thinning
+// (spec §§4–5). Safe for concurrent use; producers (exec tees, event emit
+// sites) and the ship loop share one Sink per agent process.
+type Sink struct {
+ // Budgets are fields, not constructor params, so tests and future
+ // per-app overrides set them directly; the zero Sink is unusable —
+ // NewSink applies the spec defaults.
+ MaxBytesPerApp int // ring cap per app (default 4 MiB)
+ RatePerMin float64 // sustained app-line budget per app (default 1 MiB/min)
+ Burst float64 // token bucket depth (default 4 MiB)
+
+ box string
+ now func() time.Time
+
+ mu sync.Mutex
+ apps map[AppKey]*appBuf
+ seq uint64 // per box process, shared across apps — the follow cursor's spine
+}
+
+type appBuf struct {
+ recs []Record
+ bytes int
+ tokens float64
+ lastRef time.Time
+ dropped int
+}
+
+// NewSink returns a Sink stamping records with box and timestamps from now
+// (a seam so tests control the clock).
+func NewSink(box string, now func() time.Time) *Sink {
+ return &Sink{
+ MaxBytesPerApp: 4 << 20,
+ RatePerMin: 1 << 20,
+ Burst: 4 << 20,
+ box: box,
+ now: now,
+ apps: make(map[AppKey]*appBuf),
+ }
+}
+
+// App buffers one line of app output for (acct, app), parsing structured
+// JSON lines at the door (ParseAppLine) so the bucket holds fields and
+// honored levels rather than opaque text. Subject to both budgets.
+func (s *Sink) App(acct, app, host, line string) {
+ msg, level, fields := ParseAppLine(line)
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ b := s.buf(AppKey{acct, app})
+ cost := float64(recCost(msg))
+ b.refill(s.now(), s.RatePerMin, s.Burst)
+ if b.tokens < cost {
+ b.dropped++
+ return
+ }
+ b.tokens -= cost
+ s.append(b, Record{
+ Host: host,
+ Stream: StreamApp,
+ Level: level,
+ Msg: msg,
+ Fields: fields,
+ })
+}
+
+// Event buffers one platform event — member-safe text composed at the emit
+// site (spec §5), never a relayed journal line. Events bypass the rate cap:
+// they are the diagnosis, and a chatty app must not crowd out its own
+// wake-failure line. The ring cap still applies (a platform-event storm is
+// a bug, not a budget).
+func (s *Sink) Event(acct, app, host, level, msg string) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ s.append(s.buf(AppKey{acct, app}), Record{
+ Host: host,
+ Stream: StreamPlatform,
+ Level: level,
+ Msg: msg,
+ })
+}
+
+// Drain returns everything buffered, keyed by app, and resets the buffers
+// (sequence numbers and rate-bucket state persist — they are the process's,
+// not the batch's). An app whose lines were dropped since the last drain
+// gets one trailing synthetic record naming the count. Empty apps are
+// omitted; an empty sink drains to an empty map.
+func (s *Sink) Drain() map[AppKey][]Record {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ out := make(map[AppKey][]Record)
+ for k, b := range s.apps {
+ if b.dropped > 0 {
+ s.append(b, Record{
+ Stream: StreamPlatform,
+ Level: LevelWarn,
+ Msg: fmt.Sprintf("dropped %d app log line(s) (rate cap)", b.dropped),
+ })
+ b.dropped = 0
+ }
+ if len(b.recs) == 0 {
+ continue
+ }
+ out[k] = b.recs
+ b.recs = nil
+ b.bytes = 0
+ }
+ return out
+}
+
+// buf returns (creating if needed) the app's buffer. Callers hold s.mu.
+func (s *Sink) buf(k AppKey) *appBuf {
+ b := s.apps[k]
+ if b == nil {
+ b = &appBuf{tokens: s.Burst, lastRef: s.now()}
+ s.apps[k] = b
+ }
+ return b
+}
+
+// append stamps and stores one record, evicting oldest-first past the ring
+// cap. Callers hold s.mu.
+func (s *Sink) append(b *appBuf, r Record) {
+ s.seq++
+ r.TS = s.now().UTC()
+ r.Box = s.box
+ r.Seq = s.seq
+ b.recs = append(b.recs, r)
+ b.bytes += recCost(r.Msg)
+ for b.bytes > s.MaxBytesPerApp && len(b.recs) > 1 {
+ b.bytes -= recCost(b.recs[0].Msg)
+ b.recs = b.recs[1:]
+ b.dropped++
+ }
+}
+
+// refill advances the token bucket to now. Callers hold s.mu.
+func (b *appBuf) refill(now time.Time, perMin, burst float64) {
+ elapsed := now.Sub(b.lastRef)
+ b.lastRef = now
+ b.tokens += perMin * elapsed.Minutes()
+ if b.tokens > burst {
+ b.tokens = burst
+ }
+}
+
+// recCost is the budget cost of one record: its message plus a flat
+// overhead standing in for the envelope. Both budgets use the same measure
+// so "4 MiB of ring" and "1 MiB/min of rate" describe the same bytes.
+func recCost(msg string) int { return len(msg) + 64 }
diff --git a/internal/applog/sink_test.go b/internal/applog/sink_test.go
new file mode 100644
index 000000000..da1558988
--- /dev/null
+++ b/internal/applog/sink_test.go
@@ -0,0 +1,136 @@
+package applog
+
+import (
+ "strings"
+ "testing"
+ "time"
+)
+
+func testClock(start time.Time) (func() time.Time, func(d time.Duration)) {
+ now := start
+ return func() time.Time { return now }, func(d time.Duration) { now = now.Add(d) }
+}
+
+func TestSinkStampsAndSequences(t *testing.T) {
+ now, _ := testClock(time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC))
+ sk := NewSink("box1", now)
+ sk.App("bde", "lpay", "lpay.bde.oncarlos.com", "hello")
+ sk.Event("bde", "lpay", "lpay.bde.oncarlos.com", LevelError, "wake failed: seed timed out")
+
+ got := sk.Drain()
+ if len(got) != 1 {
+ t.Fatalf("apps drained: %d, want 1", len(got))
+ }
+ recs := got[AppKey{"bde", "lpay"}]
+ if len(recs) != 2 {
+ t.Fatalf("records: %d, want 2", len(recs))
+ }
+ if recs[0].Box != "box1" || recs[0].Stream != StreamApp || recs[0].TS.IsZero() || recs[0].Host != "lpay.bde.oncarlos.com" {
+ t.Fatalf("app record not stamped: %+v", recs[0])
+ }
+ if recs[1].Stream != StreamPlatform || recs[1].Level != LevelError {
+ t.Fatalf("event record wrong: %+v", recs[1])
+ }
+ if recs[1].Seq != recs[0].Seq+1 {
+ t.Fatalf("seq not monotonic: %d then %d", recs[0].Seq, recs[1].Seq)
+ }
+ // Drain resets; seq does not.
+ sk.App("bde", "lpay", "h", "again")
+ recs2 := sk.Drain()[AppKey{"bde", "lpay"}]
+ if len(recs2) != 1 || recs2[0].Seq != recs[1].Seq+1 {
+ t.Fatalf("after drain: %+v", recs2)
+ }
+}
+
+// TestSinkAppLinesCarryParsedJSON: the sink is where ParseAppLine is
+// applied, so a structured app line arrives in the bucket with fields and
+// an honored level rather than as opaque text.
+func TestSinkAppLinesCarryParsedJSON(t *testing.T) {
+ now, _ := testClock(time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC))
+ sk := NewSink("box1", now)
+ sk.App("bde", "lpay", "h", `{"level":"warn","msg":"slow","request_id":"r1"}`)
+ rec := sk.Drain()[AppKey{"bde", "lpay"}][0]
+ if rec.Msg != "slow" || rec.Level != LevelWarn || rec.Fields["request_id"] != "r1" {
+ t.Fatalf("structured line not parsed at the sink: %+v", rec)
+ }
+}
+
+// TestSinkRingCapDropsOldestAndSaysSo: over the per-app byte cap the oldest
+// lines go first, and the drain carries one synthetic platform record
+// naming the count — never silent thinning (spec §4).
+func TestSinkRingCapDropsOldestAndSaysSo(t *testing.T) {
+ now, _ := testClock(time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC))
+ sk := NewSink("box1", now)
+ sk.MaxBytesPerApp = 600 // tiny for the test
+ sk.Burst = 1 << 20 // rate cap out of the way
+ for i := 0; i < 50; i++ {
+ sk.App("bde", "lpay", "h", strings.Repeat("x", 40))
+ }
+ recs := sk.Drain()[AppKey{"bde", "lpay"}]
+ if len(recs) >= 50 {
+ t.Fatalf("nothing evicted: %d records", len(recs))
+ }
+ last := recs[len(recs)-1]
+ if last.Stream != StreamPlatform || last.Level != LevelWarn ||
+ !strings.Contains(last.Msg, "dropped") {
+ t.Fatalf("no honest drop record; last = %+v", last)
+ }
+ // The synthetic record is accounted once per drain, not once per drop.
+ for _, r := range recs[:len(recs)-1] {
+ if strings.Contains(r.Msg, "dropped") {
+ t.Fatalf("drop record duplicated mid-stream: %+v", r)
+ }
+ }
+}
+
+// TestSinkRateCapRefills: the token bucket drops lines while exhausted and
+// admits again after time passes; drops surface in the same honest record.
+func TestSinkRateCapRefills(t *testing.T) {
+ now, advance := testClock(time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC))
+ sk := NewSink("box1", now)
+ sk.RatePerMin = 300
+ sk.Burst = 300
+ sk.App("bde", "lpay", "h", strings.Repeat("a", 200)) // fits burst
+ sk.App("bde", "lpay", "h", strings.Repeat("b", 200)) // bucket empty → dropped
+ recs := sk.Drain()[AppKey{"bde", "lpay"}]
+ if len(recs) != 2 { // the admitted line + the drop notice
+ t.Fatalf("got %d records: %+v", len(recs), recs)
+ }
+ if !strings.Contains(recs[1].Msg, "dropped 1") {
+ t.Fatalf("drop notice wrong: %+v", recs[1])
+ }
+ advance(time.Minute)
+ sk.App("bde", "lpay", "h", strings.Repeat("c", 200))
+ if got := sk.Drain()[AppKey{"bde", "lpay"}]; len(got) != 1 || !strings.HasPrefix(got[0].Msg, "ccc") {
+ t.Fatalf("bucket did not refill: %+v", got)
+ }
+}
+
+// TestSinkEventsBypassTheRateCap: platform events are the diagnosis — a
+// chatty app must not be able to crowd out its own wake-failure line.
+func TestSinkEventsBypassTheRateCap(t *testing.T) {
+ now, _ := testClock(time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC))
+ sk := NewSink("box1", now)
+ sk.RatePerMin = 10
+ sk.Burst = 10
+ sk.App("bde", "lpay", "h", strings.Repeat("a", 400)) // exhausts the bucket
+ sk.Event("bde", "lpay", "h", LevelError, "wake failed")
+ recs := sk.Drain()[AppKey{"bde", "lpay"}]
+ var sawEvent bool
+ for _, r := range recs {
+ if r.Stream == StreamPlatform && r.Msg == "wake failed" {
+ sawEvent = true
+ }
+ }
+ if !sawEvent {
+ t.Fatalf("event lost to the app's own rate cap: %+v", recs)
+ }
+}
+
+func TestSinkDrainEmptyIsEmpty(t *testing.T) {
+ now, _ := testClock(time.Now())
+ sk := NewSink("box1", now)
+ if got := sk.Drain(); len(got) != 0 {
+ t.Fatalf("empty sink drained %v", got)
+ }
+}