| 1 | // Package applog is the pure half of per-app log shipping (`carlos logs`, |
| 2 | // spec 2026-08-10-carlos-logs-design.md): the wire record, the box-side |
| 3 | // sink (ring + budgets + drop honesty), the gzipped-JSONL chunk codec and |
| 4 | // bucket key builders, and the read-side filter/merge fold. No I/O and no |
| 5 | // clock of its own — internal/host owns the shipping loop, internal/console |
| 6 | // owns the read path, exactly the internal/metrics split. |
| 7 | package applog |
| 8 | |
| 9 | import ( |
| 10 | "encoding/json" |
| 11 | "strconv" |
| 12 | "strings" |
| 13 | "time" |
| 14 | ) |
| 15 | |
| 16 | // Streams a Record can belong to. The vocabulary is fixed by the spec (§1) |
| 17 | // so later stages are additive: stage 1 ships app + platform; edge and |
| 18 | // release exist in the vocabulary now so nothing re-parses when they arrive. |
| 19 | const ( |
| 20 | StreamApp = "app" // the app process's own stdout/stderr |
| 21 | StreamPlatform = "platform" // wake/restart/lifecycle events, member-safe text |
| 22 | StreamEdge = "edge" // per-request access lines (stage 2) |
| 23 | StreamRelease = "release" // ship/promote history join (stage 3) |
| 24 | ) |
| 25 | |
| 26 | // ValidStream reports whether s names a known stream — the CLI/API's |
| 27 | // vocabulary check, kept here so there is exactly one list. |
| 28 | func ValidStream(s string) bool { |
| 29 | switch s { |
| 30 | case StreamApp, StreamPlatform, StreamEdge, StreamRelease: |
| 31 | return true |
| 32 | } |
| 33 | return false |
| 34 | } |
| 35 | |
| 36 | // Levels a Record may carry. Order matters to Filter's threshold semantics; |
| 37 | // levelRank is the one place that order lives. |
| 38 | const ( |
| 39 | LevelDebug = "debug" |
| 40 | LevelInfo = "info" |
| 41 | LevelWarn = "warn" |
| 42 | LevelError = "error" |
| 43 | ) |
| 44 | |
| 45 | // levelRank maps a level to its severity order. Unknown/absent levels rank |
| 46 | // -1: a record that never claimed a level can never satisfy a -level |
| 47 | // threshold, because a guessed level is a confident wrong answer (spec §4). |
| 48 | func levelRank(level string) int { |
| 49 | switch level { |
| 50 | case LevelDebug: |
| 51 | return 0 |
| 52 | case LevelInfo: |
| 53 | return 1 |
| 54 | case LevelWarn: |
| 55 | return 2 |
| 56 | case LevelError: |
| 57 | return 3 |
| 58 | } |
| 59 | return -1 |
| 60 | } |
| 61 | |
| 62 | // ValidLevel reports whether s names a known level. |
| 63 | func ValidLevel(s string) bool { return levelRank(s) >= 0 } |
| 64 | |
| 65 | // Record is one log line on the wire — one JSONL line inside a chunk. |
| 66 | // Additive shape (AGENTS.md rule): every field but TS and Msg is omitempty, |
| 67 | // unknown keys are ignored on read, and anything written before a shape |
| 68 | // change must still parse. TS is the box clock in UTC; cross-box order is |
| 69 | // best-effort by TS with (Box, Seq) as tiebreak — per-box order is the only |
| 70 | // promise. Seq is monotonic per box process, which is what makes the follow |
| 71 | // cursor exact and gaps detectable rather than invisible. |
| 72 | type Record struct { |
| 73 | TS time.Time `json:"ts"` |
| 74 | Box string `json:"box,omitempty"` |
| 75 | Host string `json:"host,omitempty"` // route host — how owners name instances |
| 76 | Stream string `json:"stream,omitempty"` |
| 77 | Level string `json:"level,omitempty"` |
| 78 | Msg string `json:"msg"` |
| 79 | Fields map[string]string `json:"fields,omitempty"` |
| 80 | Seq uint64 `json:"seq,omitempty"` |
| 81 | } |
| 82 | |
| 83 | // ParseAppLine interprets one line of app output. A line that parses as a |
| 84 | // top-level JSON object yields: fields from its scalar values (numbers and |
| 85 | // bools stringified; nested values ignored rather than flattened), msg from |
| 86 | // a string-valued "msg" or "message" key (the raw line when neither is |
| 87 | // present), and level ONLY when a "level" key carries a recognized value |
| 88 | // (case-insensitive; "warning" counts as warn) — an unrecognized level |
| 89 | // stays in fields as data. Anything else is opaque text: msg is the line, |
| 90 | // untouched. Level is never guessed from text (spec §4): a guessed level |
| 91 | // that makes -level error drop the crash line is worse than no level. |
| 92 | func ParseAppLine(line string) (msg, level string, fields map[string]string) { |
| 93 | trimmed := strings.TrimSpace(line) |
| 94 | if !strings.HasPrefix(trimmed, "{") { |
| 95 | return line, "", nil |
| 96 | } |
| 97 | var obj map[string]any |
| 98 | if err := json.Unmarshal([]byte(trimmed), &obj); err != nil { |
| 99 | return line, "", nil |
| 100 | } |
| 101 | msg = line |
| 102 | for _, key := range []string{"msg", "message"} { |
| 103 | if s, ok := obj[key].(string); ok { |
| 104 | msg = s |
| 105 | delete(obj, key) |
| 106 | break |
| 107 | } |
| 108 | } |
| 109 | if s, ok := obj["level"].(string); ok { |
| 110 | if l := normalizeLevel(s); l != "" { |
| 111 | level = l |
| 112 | delete(obj, "level") |
| 113 | } |
| 114 | } |
| 115 | for k, v := range obj { |
| 116 | var s string |
| 117 | switch v := v.(type) { |
| 118 | case string: |
| 119 | s = v |
| 120 | case bool: |
| 121 | s = strconv.FormatBool(v) |
| 122 | case float64: |
| 123 | s = strconv.FormatFloat(v, 'f', -1, 64) |
| 124 | default: // nested object/array/null — not flattened, not invented |
| 125 | continue |
| 126 | } |
| 127 | if fields == nil { |
| 128 | fields = make(map[string]string) |
| 129 | } |
| 130 | fields[k] = s |
| 131 | } |
| 132 | return msg, level, fields |
| 133 | } |
| 134 | |
| 135 | // normalizeLevel folds the common spellings onto the fixed vocabulary, |
| 136 | // answering "" for anything it does not recognize. |
| 137 | func normalizeLevel(s string) string { |
| 138 | switch strings.ToLower(s) { |
| 139 | case "debug", "trace": |
| 140 | return LevelDebug |
| 141 | case "info": |
| 142 | return LevelInfo |
| 143 | case "warn", "warning": |
| 144 | return LevelWarn |
| 145 | case "error", "err", "fatal", "panic": |
| 146 | return LevelError |
| 147 | } |
| 148 | return "" |
| 149 | } |
| 150 | |
| 151 | // truncMarker is appended to a line NewLineWriter cut at its cap. It reads |
| 152 | // as what happened rather than pretending the line ended there. |
| 153 | const truncMarker = "…(truncated)" |
| 154 | |
| 155 | // LineWriter adapts a byte stream (an exec child's stdout/stderr) into |
| 156 | // whole log lines: split on \n, blank lines dropped, over-long lines cut at |
| 157 | // max bytes with truncMarker appended and the remainder of that line |
| 158 | // discarded (not emitted as a phantom second line). Close flushes a |
| 159 | // trailing partial so a crash's last unterminated words still ship. |
| 160 | // |
| 161 | // Not safe for concurrent Write from multiple goroutines — each stream gets |
| 162 | // its own LineWriter, which is also what keeps interleaving per-stream. |
| 163 | type LineWriter struct { |
| 164 | max int |
| 165 | emit func(line string) |
| 166 | buf []byte |
| 167 | skipping bool // inside the discarded tail of an over-long line |
| 168 | } |
| 169 | |
| 170 | // NewLineWriter returns a LineWriter emitting complete lines to emit, each |
| 171 | // at most max bytes before the truncation marker. |
| 172 | func NewLineWriter(max int, emit func(line string)) *LineWriter { |
| 173 | return &LineWriter{max: max, emit: emit} |
| 174 | } |
| 175 | |
| 176 | // Write is infallible by design: a log tee must never be the reason an |
| 177 | // app's stdout write fails. |
| 178 | func (w *LineWriter) Write(p []byte) (int, error) { |
| 179 | for _, b := range p { |
| 180 | if b == '\n' { |
| 181 | w.flush() |
| 182 | continue |
| 183 | } |
| 184 | if w.skipping { |
| 185 | continue |
| 186 | } |
| 187 | w.buf = append(w.buf, b) |
| 188 | if len(w.buf) >= w.max { |
| 189 | w.emit(string(w.buf) + truncMarker) |
| 190 | w.buf = w.buf[:0] |
| 191 | w.skipping = true |
| 192 | } |
| 193 | } |
| 194 | return len(p), nil |
| 195 | } |
| 196 | |
| 197 | // Close flushes any trailing partial line. It never errors; the signature |
| 198 | // exists so a LineWriter can stand where an io.Closer is expected. |
| 199 | func (w *LineWriter) Close() error { |
| 200 | w.flush() |
| 201 | return nil |
| 202 | } |
| 203 | |
| 204 | func (w *LineWriter) flush() { |
| 205 | if w.skipping { |
| 206 | w.skipping = false |
| 207 | w.buf = w.buf[:0] |
| 208 | return |
| 209 | } |
| 210 | if len(w.buf) == 0 { |
| 211 | return |
| 212 | } |
| 213 | w.emit(string(w.buf)) |
| 214 | w.buf = w.buf[:0] |
| 215 | } |
| 216 | |
amadan