amadan

carlos / platform Public

Merge pull request #133 from carlosframework/carlos-logs

carlos logs: an app's merged log timeline with no box access (#131, #78)
Paul Campbell db76025d669e98fbb8a203ef8e788a0126e79ace parent ca7c86bparent 903faa0
30 files changed, +3107 −16
  • apimode.go +36 −0
  • docs/superpowers/specs/2026-08-10-carlos-logs-design.md +215 −0
  • internal/activator/activator.go +26 −0
  • internal/activator/backend_exec.go +59 −8
  • internal/activator/logsink.go +72 −0
  • internal/activator/logsink_test.go +122 −0
  • internal/applog/chunk.go +124 −0
  • internal/applog/chunk_test.go +120 −0
  • internal/applog/cursor.go +49 −0
  • internal/applog/cursor_test.go +35 −0
  • internal/applog/filter.go +108 −0
  • internal/applog/filter_test.go +91 −0
  • internal/applog/read.go +97 −0
  • internal/applog/read_test.go +106 −0
  • internal/applog/record.go +215 −0
  • internal/applog/record_test.go +133 −0
  • internal/applog/sink.go +195 −0
  • internal/applog/sink_test.go +136 −0
  • internal/console/apilogs.go +153 −0
  • internal/console/apilogs_test.go +183 −0
  • internal/console/authcli.go +3 −0
  • internal/console/server.go +54 −0
  • internal/console/templates/app_logs.html +50 −0
  • internal/console/templates/appnav.html +1 −0
  • internal/host/agent.go +28 −8
  • internal/host/logship.go +214 −0
  • internal/host/logship_test.go +142 −0
  • logs.go +177 −0
  • logs_e2e_test.go +159 −0
  • main.go +4 −0
diff --git a/apimode.go b/apimode.go
index ad4daf82e..6ae3887e6 100644
--- a/apimode.go
+++ b/apimode.go
@@ -32,6 +32,7 @@ import (
"time"
"github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/applog"
"github.com/carlosframework/platform/internal/caps"
"github.com/carlosframework/platform/internal/ledger"
)
@@ -1082,6 +1083,41 @@ func (s *apiSession) instanceList(ctx context.Context, a apiAcct, app string) ([
return res.Instances, res.Truncated, nil
}
+// apiLogsResponse is the console's logs answer (internal/console/apilogs.go)
+// — records are applog's own wire shape, decoded as-is.
+type apiLogsResponse struct {
+ Records []applog.Record `json:"records"`
+ Truncated bool `json:"truncated"`
+ Cursor string `json:"cursor"`
+}
+
+// logsGet reads one window (or one follow increment) of an app's merged log
+// timeline. query carries the filter params exactly as the endpoint speaks
+// them; apiPath may already hold a query (exact=sqid), hence the separator
+// dance.
+func (s *apiSession) logsGet(ctx context.Context, a apiAcct, app string, query url.Values) (apiLogsResponse, error) {
+ path := apiPath(a, app, "/logs")
+ if enc := query.Encode(); enc != "" {
+ if strings.Contains(path, "?") {
+ path += "&" + enc
+ } else {
+ path += "?" + enc
+ }
+ }
+ var res apiLogsResponse
+ status, body, err := s.getJSON(ctx, path)
+ if err != nil {
+ return res, err
+ }
+ if status != http.StatusOK {
+ return res, apiErrorFromBody(status, body)
+ }
+ if err := json.Unmarshal(body, &res); err != nil {
+ return res, fmt.Errorf("logs: unreadable response: %w", err)
+ }
+ return res, nil
+}
+
// instanceDelete retires one instance record. Idempotent server-side (the
// caller's goal is "not there"), so a 200 is the only success shape.
func (s *apiSession) instanceDelete(ctx context.Context, a apiAcct, app, host string) error {
diff --git a/docs/superpowers/specs/2026-08-10-carlos-logs-design.md b/docs/superpowers/specs/2026-08-10-carlos-logs-design.md
new file mode 100644
index 000000000..86299f55f
--- /dev/null
+++ b/docs/superpowers/specs/2026-08-10-carlos-logs-design.md
@@ -0,0 +1,215 @@
+# `carlos logs` — per-app logs with no box access
+
+2026-08-10. Closes the box-local-only debugging gap (#131, #78): a member
+whose app is failing gets exactly one string from the edge ("instance failed
+to wake") while the reason sits in a journal they cannot read. Paul's rule
+(CLAUDE.md 2026-08-08) is the bar: the command alone must complete the job
+for a user with zero infra access.
+
+## 1. Jobs, staged
+
+Three jobs, one surface, built in stages:
+
+1. **Diagnose a broken app** (stage 1, this spec's build): recent app output
+ plus platform-side reasons — wake errors, restart actions — merged into
+ one timeline.
+2. **Watch a deploy land** (stage 1): `carlos logs -f` follows live during
+ ship/restart.
+3. **Everyday observability** (stages 2–3): edge request lines, unit-tenant
+ journals, release-history join, longer retention knobs.
+
+The full stream vocabulary is fixed now so later stages are additive:
+
+| stream | contents | stage |
+|------------|---------------------------------------------------|-------|
+| `app` | the app process's own stdout/stderr | 1 (exec-backed), 2 (unit tenants via journald read) |
+| `platform` | wake attempts/failures, restart bounces, adopt/stage decisions, config sync, hibernation | 1 (agent-owned events), 2 (root adopt pass events — separate process, needs its own writer identity) |
+| `edge` | per-request access lines (host, method, path, status, duration, bytes; **no client IP** in the member-facing record) | 2 — does not exist today; new logging with its own volume cap |
+| `release` | ship/promote/adopt history | 3 — a read-time join over existing bucket records (manifests, channel pointers, tags), not a shipped stream |
+
+Static/bake apps have no process; for them the surface is `platform` +
+(later) `edge` lines only, and the CLI says so rather than showing an
+empty screen.
+
+## 2. Architecture
+
+One new pure package, `internal/applog` (record, sink, chunk codec,
+filter/merge — no I/O, no clock of its own), with thin plumbing around it.
+Deliberately the `internal/metrics` / `internal/host/metrics_loop.go` split.
+
+```
+exec child stdout/stderr ─┐
+agent platform events ────┼→ applog.Sink ─→ ship loop ─→ apps/<acct>/<app>/logs/<box>/<hour>/<seq>.jsonl.gz
+(edge lines, stage 2) ────┘ (ring, budget, │
+ drop honesty) ▼
+ console: merge boxes, filter, page
+ │
+ ▼
+ carlos logs / console app tab
+```
+
+Boxes push; the console never reaches into a box — the invariant `carlos
+restart` was built around. Works identically for customer fleets (they
+already write into the deployment bucket). Survives box death and
+hibernation: whatever shipped before the box died is readable.
+
+Transport choice (over on-demand rendezvous): rendezvous needs a *live* box
+— no logs precisely when the box is the problem — and gives no history; the
+box would need a local ring anyway. Continuous shipping serves all three
+jobs and reuses a flush loop this repo has already debugged once.
+
+## 3. Storage layout
+
+`apps/<acct>/<app>/logs/<box>/<YYYY-MM-DDTHH>/<seq>.jsonl.gz`
+
+- **Per-app, not per-box** (metrics is per-box because the host is its
+ subject). The subject here is the app and the reader is its owner, so the
+ console's per-(account, app) owner gate maps onto the prefix — no
+ fleet-wide scan per read. App trash/purge sweeps logs for free (they live
+ under `release.AppPrefix`).
+- **Hour directories**: retention is a prefix delete; `-since` skips whole
+ prefixes unlisted.
+- **One object per flush, never appended or rewritten.** No S3 append, and
+ the agent must not read-modify-write keys another writer could touch.
+- `<box>` is `status.Label(readMachineID())` — the identity metrics already
+ publishes under. `<seq>` is the first record's per-box sequence number,
+ zero-padded to 12 digits so lexical order is numeric order.
+- Chunks are gzipped JSONL, one record per line.
+
+## 4. Record shape (wire)
+
+```json
+{"ts":"2026-08-10T14:03:11.482913Z","box":"carlos-tito-ie",
+ "host":"lpay.bde.oncarlos.com","stream":"app","level":"error",
+ "msg":"bind: address already in use","fields":{"request_id":"a1b2"},
+ "seq":41822}
+```
+
+- `ts` — box clock, UTC RFC3339Nano. Cross-box order is best-effort by
+ `ts` with `(box, seq)` tiebreak; the spec promises per-box order only.
+- `host` — the route host; how owners already name instances.
+- `stream` — `app|platform|edge|release` (vocabulary fixed above).
+- `level` — set for platform/edge lines and for app lines **only when the
+ app's own JSON carried one**. Never guessed from text: a guessed level
+ that drops the crash line under `-level error` is worse than no flag.
+- `fields` — present only when the app line parsed as a flat-ish JSON
+ object; enables `-field k=v` from day one. Non-JSON output is opaque
+ `msg`, untouched.
+- `seq` — monotonic per box process; makes the follow cursor exact and
+ drops detectable.
+- Additive wire shape (repo rule): every field but `ts` and `msg` optional;
+ unknown fields ignored on read; anything written before a change must
+ still parse.
+- One synthetic record shape: when the sink drops lines (budget/rate), it
+ emits `{"stream":"platform","level":"warn","msg":"dropped N app log
+ line(s) (rate cap)"}` — never silent thinning.
+
+## 5. Box side (stage 1)
+
+**Producers.**
+
+- *Exec children and seeds* (`internal/activator/backend_exec.go`): today
+ `cmd.Stdout/Stderr = os.Stdout/os.Stderr`. Becomes a tee — journald
+ exactly as before, plus a line-splitting writer into the sink tagged
+ `(host, stream "app")`, stderr and stdout alike. Lines capped (8 KiB;
+ longer is truncated with a marker suffix). The tee is an `io.Writer`;
+ the WaitDelay discipline around piped stderr already in place covers it.
+- *Platform events*: an injected `Events` callback on the activator (and
+ the agent's restart dispatch) — **curated member-safe messages at each
+ emit site**, per #131's sanitize note, never relayed raw journal lines.
+ Stage-1 emit sites: wake start, wake failure (the member-safe reason:
+ e.g. "seed timed out", "no adoptable head for channel X", "instance
+ socket already in use"), instance exit, restart bounce, hibernation
+ stop. The callback is nil-safe so tests and non-shipping deployments
+ change nothing.
+
+**Sink** (`applog.Sink`): bounded ring per app (default 4 MiB), token-bucket
+rate cap per app (default 1 MiB/min sustained, 4 MiB burst). Over budget →
+oldest dropped first + the synthetic drop record. `Drain()` hands the ship
+loop everything and resets.
+
+**Ship loop** (`internal/host/logship.go`): 10s tick; flush when non-empty;
+platform events poke an immediate (2s-debounced) flush so `carlos deploy`'s
+timeout window can see the wake error. Same panic-recover + disabled-without-
+bucket shape as `metricsLoop`. On crossing an hour boundary it prunes **its
+own `<box>` subtree only** (clean ownership): hours older than 7 days, and
+oldest-first beyond the per-app-per-box byte cap (default 50 MiB).
+
+## 6. Console read path (stage 1)
+
+`GET /api/cli/apps/{account}/{app}/logs` — registered next to the instances
+family in `authcli.go`, behind the same gate shape (`instanceGate`'s order:
+principal → rate → bucket).
+
+Query params: `since`, `until` (RFC3339 or Go duration shorthand),
+`stream`, `host`, `level`, `grep` (RE2, applied console-side), `field`
+(`k=v`, repeatable), `limit` (default 1000, max 5000), `cursor`.
+
+Read: `ListDirs(logs/)` → boxes; per box, hour prefixes clipped to
+[since, until]; decode chunks newest-last; filter; k-way merge by
+`(ts, box, seq)`; cap by `limit` **and** a byte ceiling (2 MiB) with a
+`truncated` flag and an opaque `cursor` (base64 of last `(ts, box, seq)`)
+— truncation is stated, never implied. Response:
+
+```json
+{"records":[…],"truncated":false,"cursor":"…"}
+```
+
+`-f` is the same endpoint polled at 2s with the returned cursor; no
+server-side streaming, no new connection discipline.
+
+## 7. CLI (stage 1)
+
+`carlos logs -app <app> [-account <acct>] [-since 1h] [-until …]
+[-stream app|platform|edge|release] [-host <instance>] [-level warn|error]
+[-grep <re>] [-field k=v] [-n <limit>] [-f]`
+
+Remote form only (console API; account/app resolution and config defaults
+exactly as `carlos instances list`). Default window: last 1h, last 1000
+lines, honest "showing last N — use -since for more" when capped. Render:
+`15:04:05 [stream host] msg`, level colored via the existing color
+conventions; `fields` appended `k=v` when present. `-f` polls at 2s until
+interrupted. A static/bake app with no process prints the platform stream
+and says why there is no `app` stream.
+
+## 8. Retention & cost
+
+7 days, per-app-per-box 50 MiB cap, oldest-first, enforced by each box for
+its own subtree. Steady-state cost: one small PUT per box per 10s *only
+when something logged*; an idle box ships nothing. When either cap binds,
+the drop is recorded in-stream (synthetic record) and visible in the CLI.
+
+## 9. Security
+
+- Owner-gated end to end: the API resolves the account exactly like the
+ instances family; a non-member gets the byte-identical 404.
+- Platform-event text is composed member-safe at the emit site (no box
+ paths, no journal relay). App stdout is the owner's own output and ships
+ verbatim (minus line-length caps).
+- Edge lines (stage 2) will carry no client IP in the member-facing record.
+- The bucket prefix lives under the app, so deletion policy and trash
+ already govern it.
+
+## 10. Testing
+
+- `internal/applog`: record round-trip + additive-parse pinning, sink
+ budget/drop honesty, chunk codec, filter matrix, merge order.
+- Activator: tee preserves journald bytes; child lines land tagged; wake
+ failure emits exactly one member-safe platform event (fixture Events fn).
+- Ship loop: DirStore-backed flush/prune/no-empty-PUT.
+- Console: handler filter/paging/gate tests on a DirStore fixture.
+- CLI: e2e against a fixture console via `carlosTestEnv` + scratch
+ `cmd.Dir` (the repo's minted-real-accounts rule).
+- The full repo gate: build/vet/test every package, `GOOS=darwin` compile.
+
+## 11. Out of scope (recorded so they stay decisions, not accidents)
+
+- Root adopt pass skip reasons in-stream (stage 2; the other half of #78 —
+ it is a separate process needing its own writer identity/seq).
+- Unit-tenant journald ingestion (stage 2).
+- Edge request lines (stage 2) and the release join (stage 3).
+- Full-text index/search over all history; alerting on log content.
+
+(The console web UI Logs tab on the app page IS stage 1 — Paul, 2026-08-10:
+a member should see logs in the console, not only through the CLI. It
+renders the same merged timeline server-side with stream/since filters.)
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 9014bfe37..9a07ae907 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -363,6 +363,12 @@ type Service struct {
now func() time.Time // a seam for the unit tests' clock
liveVer func(string) (string, error) // live/<host> symlink target; a seam for tests
+ // Logs, when non-nil, receives member-visible platform events (wake
+ // outcomes, restart bounces) for `carlos logs` — set before Start, like
+ // the backend's own Logs field. Nil (every test, a box without the
+ // shipping loop) emits nothing.
+ Logs EventSink
+
// mesh seam: Eleven kept the multi-box placement layer here (mesh *meshSvc):
// requests for hosts whose HRW owner is another live box were proxied
// there, and the due-time sweeper only woke hosts this box owned. Nil
@@ -573,6 +579,19 @@ func (a *Service) wake(e *actEntry) {
defer cancel()
w, err := a.doWake(ctx, e.route)
+ // The member-facing record of this wake (spec §5): composed here, where
+ // the route and the whole error are in hand — the journal line below the
+ // backends already write stays exactly as it was.
+ if err != nil {
+ emitEvent(a.Logs, e.route, "error", memberSafeWakeReason(err))
+ } else {
+ msg := "instance woke"
+ if w.spawnVersion != "" {
+ msg += " (version " + w.spawnVersion + ")"
+ }
+ emitEvent(a.Logs, e.route, "info", msg)
+ }
+
a.mu.Lock()
if err != nil {
e.err = err
@@ -737,6 +756,7 @@ func (a *Service) Bounce(host string) bool {
if e == nil || e.state != actAwake {
return false
}
+ emitEvent(a.Logs, e.route, "info", "instance restarting (carlos restart)")
a.beginStopLocked(e)
return true
}
@@ -746,6 +766,12 @@ func (a *Service) Bounce(host string) bool {
// the entry. The local DB stays in place — it's the warm cache for the next
// wake; S3 is the durable truth under the lease.
func (a *Service) stop(e *actEntry) {
+ // The deliberate-stop marker for the member's timeline: hibernation,
+ // convergence cycling, bounce, shutdown all pass through here, so an
+ // "exited" line WITHOUT this line before it reads as what it is — a
+ // crash.
+ emitEvent(a.Logs, e.route, "info", "instance stopping")
+
// Co-stop, sidecar FIRST — before the next-due probe and the instance's
// own SIGTERM. Two reasons: the sidecar must never outlive its instance
// (an orphan's reconnect loop would be a standing wake source — the
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 57775d560..22c8d30ce 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -20,6 +20,7 @@ import (
"syscall"
"time"
+ "github.com/carlosframework/platform/internal/applog"
"github.com/carlosframework/platform/internal/cgroup"
"github.com/carlosframework/platform/internal/edge"
"github.com/carlosframework/platform/internal/lease"
@@ -62,6 +63,40 @@ type ExecBackend struct {
// metrics sampler can account it (exec-child accounting spec). Nil — no
// delegation: old unit file, tests, darwin — spawns exactly as before.
Cgroups *cgroup.Manager
+ // Logs, when non-nil, gets each instance and seed child's stdout/stderr
+ // teed in line-wise for `carlos logs` (spec §5) — journald keeps every
+ // byte it got before. Nil (tests, a box without the shipping loop)
+ // spawns exactly as before: fd passthrough, no pipes.
+ Logs EventSink
+}
+
+// execLogLineMax caps one teed log line (spec §5): longer is truncated with
+// a marker, never split into phantom lines.
+const execLogLineMax = 8 << 10
+
+// execDrainGrace bounds how long a dead child's held-open output pipes can
+// delay its Wait. Teeing turns the child's stdout/stderr into PIPES (with a
+// plain *os.File the fd passed straight through), and os/exec's Wait then
+// waits for the read side to drain — which a grandchild holding the write
+// end keeps open forever. Same discipline, same reason as seedDrainGrace;
+// without it, an exited instance would look alive to the activator for as
+// long as some stray grandchild lived, and Stop would hang on p.done.
+var execDrainGrace = 5 * time.Second
+
+// teeStd wires cmd's stdout/stderr: straight to the agent's own streams
+// (journald under systemd) exactly as before, PLUS — only when b.Logs is
+// set — a line-splitting tee into the log sink tagged as rt's app output.
+func (b *ExecBackend) teeStd(cmd *exec.Cmd, rt registry.Route) {
+ if b.Logs == nil {
+ cmd.Stdout = os.Stdout
+ cmd.Stderr = os.Stderr
+ return
+ }
+ acct, app := release.RouteAccount(rt), rt.App
+ emit := func(line string) { b.Logs.App(acct, app, rt.Host, line) }
+ cmd.Stdout = io.MultiWriter(os.Stdout, applog.NewLineWriter(execLogLineMax, emit))
+ cmd.Stderr = io.MultiWriter(os.Stderr, applog.NewLineWriter(execLogLineMax, emit))
+ cmd.WaitDelay = execDrainGrace
}
var _ Backend = (*ExecBackend)(nil)
@@ -238,9 +273,17 @@ func (b *ExecBackend) Seed(ctx context.Context, rt registry.Route) error {
// Like the instance child, the seed logs through the agent's own streams
// (journald under systemd) — and its stderr is ALSO teed into a bounded
// buffer, because the wake error this returns is the whole diagnosis an
- // operator gets for a host that refuses to come up.
+ // operator gets for a host that refuses to come up. With Logs set the
+ // seed's output ships to `carlos logs` too: it is app output, and on a
+ // failed first wake it is the diagnosis the MEMBER gets.
cmd.Stdout = os.Stdout
cmd.Stderr = io.MultiWriter(os.Stderr, &stderr)
+ if b.Logs != nil {
+ acct, app := release.RouteAccount(rt), rt.App
+ emit := func(line string) { b.Logs.App(acct, app, host, line) }
+ cmd.Stdout = io.MultiWriter(os.Stdout, applog.NewLineWriter(execLogLineMax, emit))
+ cmd.Stderr = io.MultiWriter(os.Stderr, &stderr, applog.NewLineWriter(execLogLineMax, emit))
+ }
// WaitDelay is what makes seedTimeout real. Because Stderr is a plain
// io.Writer (the tee), os/exec pipes it and Wait waits for that pipe to
// close — and a seed that spawned its own grandchild holds the write end
@@ -474,9 +517,10 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
cmd.Env = childEnv(childVars)
// The child logs through the agent's own stderr/stdout (journald
// under systemd); there's no per-instance unit to own a log stream
- // anymore.
- cmd.Stdout = os.Stdout
- cmd.Stderr = os.Stderr
+ // anymore. With Logs set the same bytes are ALSO teed line-wise
+ // into the log sink (`carlos logs`); teeStd owns the WaitDelay
+ // that piping entails.
+ b.teeStd(cmd, rt)
return cmd
})
if err != nil {
@@ -505,6 +549,14 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
}
go func() {
p.err = cmd.Wait()
+ // The member-facing record of the exit: a crash shows an "exited"
+ // line with no "stopping" line before it — that pairing, not the
+ // exit code alone, is what tells a member crash from lifecycle.
+ if p.err != nil {
+ emitEvent(b.Logs, rt, "warn", "instance exited: "+redactPaths(p.err.Error()))
+ } else {
+ emitEvent(b.Logs, rt, "info", "instance exited (exit 0)")
+ }
// Last exiter wins: rmdir refuses while the sidecar (same host
// cgroup) is still alive, and that failure is tolerated.
b.Cgroups.CleanupHost(host)
@@ -668,10 +720,9 @@ func (b *ExecBackend) StartSidecar(_ context.Context, rt registry.Route) (Proc,
cmd := exec.Command(b.liveLink(rt.Host), "sidecar", "run")
cmd.Env = childEnv(fileEnv)
// Like the instance child: logs through the agent's own streams
- // (journald under systemd) — no per-host unit owns a log stream
- // anymore.
- cmd.Stdout = os.Stdout
- cmd.Stderr = os.Stderr
+ // (journald under systemd), teed into the log sink when armed — a
+ // sidecar's output is the app's output.
+ b.teeStd(cmd, rt)
return cmd
})
if err != nil {
diff --git a/internal/activator/logsink.go b/internal/activator/logsink.go
new file mode 100644
index 000000000..e8fa84242
--- /dev/null
+++ b/internal/activator/logsink.go
@@ -0,0 +1,72 @@
+// The activator's half of `carlos logs` (spec 2026-08-10-carlos-logs-design
+// §5): the seam the agent injects its applog sink through, and the
+// member-safe rendering of wake failures. The reason a wake failed is the
+// whole diagnosis a member gets for a 503 (#131), and it is composed HERE,
+// at the emit site — never relayed from the journal, and never carrying box
+// filesystem layout.
+package activator
+
+import (
+ "regexp"
+ "strings"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// EventSink is where the activator publishes member-visible log lines.
+// applog.Sink satisfies it; the interface lives here so this package
+// depends on the contract, not the buffer. Nil disables everything —
+// tests and non-shipping deployments change nothing.
+type EventSink interface {
+ // App records one line of the app's own output.
+ App(acct, app, host, line string)
+ // Event records one platform event with member-safe text.
+ Event(acct, app, host, level, msg string)
+}
+
+// emitEvent publishes one platform event for rt's app, tolerating a nil
+// sink. level is an applog.Level* value ("info"/"warn"/"error" — string
+// here to keep the dependency one-way).
+func emitEvent(sink EventSink, rt registry.Route, level, msg string) {
+ if sink == nil {
+ return
+ }
+ sink.Event(release.RouteAccount(rt), rt.App, rt.Host, level, msg)
+}
+
+// absPathToken matches an absolute-path token: box filesystem layout,
+// which a member-safe message must not carry (#131). Replacement keeps the
+// base name — "…/lpay.bde.oncarlos.com.sock" still tells the member WHICH
+// socket without telling them where the box keeps it.
+var absPathToken = regexp.MustCompile(`(^|[\s"(=])(/[\w@.\-/]+)`)
+
+// memberSafeWakeReason renders a wake error for the member-facing platform
+// stream. Known failure shapes get a diagnosis in the member's terms; the
+// rest pass through with absolute paths reduced to base names. The error
+// text itself originates in our own code (backends, materialize, lease) —
+// this is a rendering pass, not a secrecy boundary.
+func memberSafeWakeReason(err error) string {
+ s := err.Error()
+ switch {
+ case strings.Contains(s, "address already in use"):
+ // The stale-socket wake trap (#129): every later wake fails until
+ // the leftover socket goes away. Say that, not the bind syscall.
+ return "wake failed: instance socket already in use — a previous run's socket was left behind (this wake will keep failing until the platform clears it)"
+ case strings.Contains(s, "no adoptable head"):
+ return "wake failed: nothing is promoted for this instance's channel (or the head is still bake-held) — promote a build, or check `carlos channels`"
+ }
+ return "wake failed: " + redactPaths(s)
+}
+
+// redactPaths reduces absolute-path tokens to their base names.
+func redactPaths(s string) string {
+ return absPathToken.ReplaceAllStringFunc(s, func(m string) string {
+ i := strings.Index(m, "/")
+ prefix, path := m[:i], m[i:]
+ if j := strings.LastIndexByte(path, '/'); j >= 0 && j+1 < len(path) {
+ return prefix + path[j+1:]
+ }
+ return prefix + path
+ })
+}
diff --git a/internal/activator/logsink_test.go b/internal/activator/logsink_test.go
new file mode 100644
index 000000000..c119cee1e
--- /dev/null
+++ b/internal/activator/logsink_test.go
@@ -0,0 +1,122 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "strings"
+ "testing"
+ "time"
+)
+
+// TestMemberSafeWakeReason pins the two real 2026-08-09 cases from #131 —
+// each needs a different response from the member, and neither may carry
+// box filesystem layout.
+func TestMemberSafeWakeReason(t *testing.T) {
+ cases := []struct {
+ name string
+ err error
+ want []string // substrings that must appear
+ deny []string // substrings that must NOT appear
+ }{
+ {
+ "stale socket (#129)",
+ errors.New(`start lpay.bde.oncarlos.com: listen unix /data/exec/lpay.bde.oncarlos.com/inst.sock: bind: address already in use`),
+ []string{"socket already in use", "previous run"},
+ []string{"/data", "bind:"},
+ },
+ {
+ "nothing promoted (#130)",
+ errors.New(`materialize lpay.bde.oncarlos.com: no adoptable head (nothing promoted for its channel, or the head is bake-held)`),
+ []string{"nothing is promoted", "carlos channels"},
+ nil,
+ },
+ {
+ "unknown errors keep meaning, lose layout",
+ errors.New(`seed h: fork/exec /opt/carlos/versions/lpay/1.2.3/lpay: permission denied`),
+ []string{"wake failed", "permission denied", "lpay"},
+ []string{"/opt/carlos"},
+ },
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ got := memberSafeWakeReason(tc.err)
+ for _, w := range tc.want {
+ if !strings.Contains(got, w) {
+ t.Errorf("missing %q in %q", w, got)
+ }
+ }
+ for _, d := range tc.deny {
+ if strings.Contains(got, d) {
+ t.Errorf("leaked %q in %q", d, got)
+ }
+ }
+ })
+ }
+}
+
+// eventRec and fakeSink are the package's EventSink fixture.
+type eventRec struct{ acct, app, host, level, msg string }
+
+type fakeSink struct {
+ apps []eventRec
+ events []eventRec
+}
+
+func (f *fakeSink) App(acct, app, host, line string) {
+ f.apps = append(f.apps, eventRec{acct, app, host, "", line})
+}
+func (f *fakeSink) Event(acct, app, host, level, msg string) {
+ f.events = append(f.events, eventRec{acct, app, host, level, msg})
+}
+
+// TestWakeOutcomesReachTheSink: the activator's wake path publishes exactly
+// one member-facing event per outcome — the #131 fact, at its source. The
+// route carries app+account context so the sink can key the record.
+func TestWakeOutcomesReachTheSink(t *testing.T) {
+ // Failure: Start refuses, the event is the member-safe rendering.
+ a, b, _, _ := testActivator(t, time.Hour)
+ sink := &fakeSink{}
+ a.Logs = sink
+ b.startErr = errors.New("listen unix /data/exec/h/inst.sock: bind: address already in use")
+ rt := testRoute("lpay.bde.oncarlos.com")
+ rt.App = "lpay"
+ rt.Slug = ""
+ if _, err := a.beginRequest(context.Background(), rt); err == nil {
+ t.Fatal("wake should have failed")
+ }
+ var wakeFails []eventRec
+ for _, e := range sink.events {
+ if strings.Contains(e.msg, "wake failed") {
+ wakeFails = append(wakeFails, e)
+ }
+ }
+ if len(wakeFails) != 1 {
+ t.Fatalf("wake-failure events: %d (%v)", len(wakeFails), sink.events)
+ }
+ ev := wakeFails[0]
+ if ev.level != "error" || ev.host != "lpay.bde.oncarlos.com" || ev.app != "lpay" {
+ t.Fatalf("event mis-tagged: %+v", ev)
+ }
+ if strings.Contains(ev.msg, "/data") || !strings.Contains(ev.msg, "socket already in use") {
+ t.Fatalf("event not member-safe: %q", ev.msg)
+ }
+
+ // Success: a woke event, tagged the same way.
+ a2, _, _, _ := testActivator(t, time.Hour)
+ sink2 := &fakeSink{}
+ a2.Logs = sink2
+ rel, err := a2.beginRequest(context.Background(), rt)
+ if err != nil {
+ t.Fatalf("wake failed: %v", err)
+ }
+ rel()
+ var woke bool
+ for _, e := range sink2.events {
+ if strings.Contains(e.msg, "instance woke") && e.level == "info" {
+ woke = true
+ }
+ }
+ if !woke {
+ t.Fatalf("no woke event: %+v", sink2.events)
+ }
+}
diff --git a/internal/applog/chunk.go b/internal/applog/chunk.go
new file mode 100644
index 000000000..6603f8ab0
--- /dev/null
+++ b/internal/applog/chunk.go
@@ -0,0 +1,124 @@
+package applog
+
+import (
+ "bufio"
+ "bytes"
+ "compress/gzip"
+ "encoding/json"
+ "fmt"
+ "io"
+ "strconv"
+ "strings"
+ "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 (16 digits — seq is milliseconds-based, see
+// NewSink) so lexical listing order is numeric order, and size is the
+// object's own byte length. Size lives IN the key on purpose:
+// release.Store's List returns keys only, and retention's byte cap needs
+// sizes — embedding them costs a suffix, extending the Store interface
+// costs every implementation. Chunks are written once and never appended
+// or rewritten (spec §3).
+func ChunkKey(acct, app, box, hour string, seq uint64, size int) string {
+ return fmt.Sprintf("%s%016d-%d.jsonl.gz", HourPrefix(acct, app, box, hour), seq, size)
+}
+
+// ChunkKeySize recovers the byte size a ChunkKey carries, false for any key
+// not shaped like one (retention treats unparsable as size-unknown = 0 —
+// age still prunes it, the byte cap just cannot count it).
+func ChunkKeySize(key string) (int64, bool) {
+ base := key
+ if i := strings.LastIndexByte(base, '/'); i >= 0 {
+ base = base[i+1:]
+ }
+ base, ok := strings.CutSuffix(base, ".jsonl.gz")
+ if !ok {
+ return 0, false
+ }
+ _, sizeStr, ok := strings.Cut(base, "-")
+ if !ok {
+ return 0, false
+ }
+ n, err := strconv.ParseInt(sizeStr, 10, 64)
+ if err != nil || n < 0 {
+ return 0, false
+ }
+ return n, true
+}
+
+// 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..e5069f123
--- /dev/null
+++ b/internal/applog/chunk_test.go
@@ -0,0 +1,120 @@
+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; the byte size rides in
+ // the key so retention can count without a Store interface change.
+ key := ChunkKey("bde", "lpay", "box1", "2026-08-10-21", 41822, 8123)
+ if key != "apps/bde/lpay/logs/box1/2026-08-10-21/0000000000041822-8123.jsonl.gz" {
+ t.Fatalf("ChunkKey = %q", key)
+ }
+ if n, ok := ChunkKeySize(key); !ok || n != 8123 {
+ t.Fatalf("ChunkKeySize = %d, %v", n, ok)
+ }
+ for _, bad := range []string{"x/no-size.jsonl.gz", "x/whatever.json", "0000000000041822-8123.jsonl.gz.bak"} {
+ if _, ok := ChunkKeySize(bad); ok {
+ t.Errorf("ChunkKeySize(%q) parsed", bad)
+ }
+ }
+}
+
+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/cursor.go b/internal/applog/cursor.go
new file mode 100644
index 000000000..3748ca849
--- /dev/null
+++ b/internal/applog/cursor.go
@@ -0,0 +1,49 @@
+package applog
+
+import (
+ "encoding/base64"
+ "encoding/json"
+ "time"
+)
+
+// Cursor names a position in the merged timeline — the last record a reader
+// has seen, by exactly the (TS, Box, Seq) triple MergeRecords orders by.
+// Opaque base64 on the wire so its shape is the server's to evolve.
+type Cursor struct {
+ TS time.Time `json:"ts"`
+ Box string `json:"box"`
+ Seq uint64 `json:"seq"`
+}
+
+// Before reports whether c sits strictly before r in merge order — the
+// follow test: records with c.Before(r) are new since the cursor.
+func (c Cursor) Before(r Record) bool {
+ if !c.TS.Equal(r.TS) {
+ return c.TS.Before(r.TS)
+ }
+ if c.Box != r.Box {
+ return c.Box < r.Box
+ }
+ return c.Seq < r.Seq
+}
+
+// EncodeCursor renders the position just after r.
+func EncodeCursor(r Record) string {
+ b, _ := json.Marshal(Cursor{TS: r.TS, Box: r.Box, Seq: r.Seq}) // cannot fail on this shape
+ return base64.RawURLEncoding.EncodeToString(b)
+}
+
+// DecodeCursor parses what EncodeCursor produced; ok=false for anything
+// else (including future shapes this build cannot read — the caller treats
+// that as "no cursor", which re-tails rather than erroring a follow).
+func DecodeCursor(s string) (Cursor, bool) {
+ b, err := base64.RawURLEncoding.DecodeString(s)
+ if err != nil {
+ return Cursor{}, false
+ }
+ var c Cursor
+ if json.Unmarshal(b, &c) != nil || c.TS.IsZero() {
+ return Cursor{}, false
+ }
+ return c, true
+}
diff --git a/internal/applog/cursor_test.go b/internal/applog/cursor_test.go
new file mode 100644
index 000000000..433bdd818
--- /dev/null
+++ b/internal/applog/cursor_test.go
@@ -0,0 +1,35 @@
+package applog
+
+import (
+ "testing"
+ "time"
+)
+
+func TestCursorRoundTripAndOrder(t *testing.T) {
+ r := Record{TS: time.Date(2026, 8, 10, 12, 0, 30, 500, time.UTC), Box: "b1", Seq: 42, Msg: "m"}
+ c, ok := DecodeCursor(EncodeCursor(r))
+ if !ok || !c.TS.Equal(r.TS) || c.Box != "b1" || c.Seq != 42 {
+ t.Fatalf("round trip: %+v %v", c, ok)
+ }
+ // Before mirrors MergeRecords' (ts, box, seq) order exactly.
+ if c.Before(r) {
+ t.Fatal("a cursor at r is not before r")
+ }
+ for _, after := range []Record{
+ {TS: r.TS.Add(time.Nanosecond), Box: "a0", Seq: 1},
+ {TS: r.TS, Box: "b2", Seq: 1},
+ {TS: r.TS, Box: "b1", Seq: 43},
+ } {
+ if !c.Before(after) {
+ t.Fatalf("cursor should be before %+v", after)
+ }
+ }
+ if c.Before(Record{TS: r.TS, Box: "b1", Seq: 41}) {
+ t.Fatal("cursor before an earlier seq")
+ }
+ for _, bad := range []string{"", "!!!", "bm90IGpzb24"} {
+ if _, ok := DecodeCursor(bad); ok {
+ t.Errorf("DecodeCursor(%q) ok", bad)
+ }
+ }
+}
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/read.go b/internal/applog/read.go
new file mode 100644
index 000000000..d0c41ea83
--- /dev/null
+++ b/internal/applog/read.go
@@ -0,0 +1,97 @@
+package applog
+
+import (
+ "context"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// ReadWindow assembles one app's merged timeline from the bucket: every
+// box's chunks whose hour directories intersect [f.Since, f.Until], decoded,
+// filtered through f, merged in (ts, box, seq) order. Hour pruning happens
+// at the PREFIX level — an hour outside the window is never listed, which
+// is what makes a 1h read over 7d of history cheap (spec §3).
+//
+// collectCap bounds memory against a pathological window: past it the
+// OLDEST records go (the tail is what a log reader came for), and the
+// second return says so. Store errors fail the read — a partial answer
+// that looks whole is the one thing this must never produce.
+func ReadWindow(ctx context.Context, s release.Store, acct, app string, f Filter, collectCap int) ([]Record, bool, error) {
+ boxDirs, err := s.ListDirs(ctx, Prefix(acct, app))
+ if err != nil {
+ return nil, false, err
+ }
+ trimmed := false
+ perBox := make([][]Record, 0, len(boxDirs))
+ for _, boxDir := range boxDirs {
+ hourDirs, err := s.ListDirs(ctx, boxDir)
+ if err != nil {
+ return nil, false, err
+ }
+ var recs []Record
+ for _, hourDir := range hourDirs { // lexically sorted = chronological
+ if !hourIntersects(hourNameOfDir(hourDir), f.Since, f.Until) {
+ continue
+ }
+ keys, err := s.List(ctx, hourDir)
+ if err != nil {
+ return nil, false, err
+ }
+ for _, key := range keys { // lexically sorted = seq order
+ body, _, err := s.Get(ctx, key)
+ if err != nil {
+ return nil, false, err
+ }
+ chunk, err := DecodeChunk(body)
+ if err != nil {
+ return nil, false, err
+ }
+ for _, r := range chunk {
+ if f.Match(r) {
+ recs = append(recs, r)
+ }
+ }
+ if over := len(recs) - collectCap; over > 0 {
+ recs = recs[over:]
+ trimmed = true
+ }
+ }
+ }
+ perBox = append(perBox, recs)
+ }
+ merged := MergeRecords(perBox...)
+ if over := len(merged) - collectCap; over > 0 {
+ merged = merged[over:]
+ trimmed = true
+ }
+ return merged, trimmed, nil
+}
+
+// hourIntersects reports whether the hour directory named hour overlaps
+// [since, until] (zero = unbounded). Unparsable names never match — an
+// alien directory is not silently read as data.
+func hourIntersects(hour string, since, until time.Time) bool {
+ start, err := time.ParseInLocation(hourFormat, hour, time.UTC)
+ if err != nil {
+ return false
+ }
+ end := start.Add(time.Hour)
+ if !since.IsZero() && !end.After(since) {
+ return false
+ }
+ if !until.IsZero() && start.After(until) {
+ return false
+ }
+ return true
+}
+
+// hourNameOfDir extracts the final path segment of a "/"-terminated prefix.
+func hourNameOfDir(dir string) string {
+ trimmed := strings.TrimSuffix(dir, "/")
+ if i := strings.LastIndexByte(trimmed, '/'); i >= 0 {
+ return trimmed[i+1:]
+ }
+ return trimmed
+}
diff --git a/internal/applog/read_test.go b/internal/applog/read_test.go
new file mode 100644
index 000000000..17883a1a8
--- /dev/null
+++ b/internal/applog/read_test.go
@@ -0,0 +1,106 @@
+package applog
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// seedWindow writes chunks for two boxes across three hours.
+func seedWindow(t *testing.T, s release.Store) {
+ t.Helper()
+ ctx := context.Background()
+ write := func(box, hour string, seq uint64, recs []Record) {
+ t.Helper()
+ body, err := EncodeChunk(recs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := s.Put(ctx, ChunkKey("bde", "lpay", box, hour, seq, len(body)), body, ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ ts := func(h, m int) time.Time { return time.Date(2026, 8, 10, h, m, 0, 0, time.UTC) }
+ write("box1", "2026-08-10-10", 100, []Record{
+ {TS: ts(10, 5), Box: "box1", Seq: 100, Stream: StreamApp, Host: "h1", Msg: "old line"},
+ })
+ write("box1", "2026-08-10-11", 200, []Record{
+ {TS: ts(11, 10), Box: "box1", Seq: 200, Stream: StreamApp, Host: "h1", Msg: "b1 app"},
+ {TS: ts(11, 30), Box: "box1", Seq: 201, Stream: StreamPlatform, Level: LevelError, Host: "h1", Msg: "wake failed: x"},
+ })
+ write("box2", "2026-08-10-11", 500, []Record{
+ {TS: ts(11, 20), Box: "box2", Seq: 500, Stream: StreamApp, Host: "h2", Msg: "b2 app"},
+ })
+}
+
+func TestReadWindowMergesAndClipsHours(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ seedWindow(t, s)
+ ctx := context.Background()
+
+ // Window covering hour 11 only: the hour-10 chunk is never even listed.
+ f := Filter{Since: time.Date(2026, 8, 10, 11, 0, 0, 0, time.UTC)}
+ recs, trimmed, err := ReadWindow(ctx, s, "bde", "lpay", f, 1000)
+ if err != nil || trimmed {
+ t.Fatalf("read: %v trimmed=%v", err, trimmed)
+ }
+ want := []string{"b1 app", "b2 app", "wake failed: x"} // (ts, box, seq) order
+ if len(recs) != len(want) {
+ t.Fatalf("got %d records: %+v", len(recs), recs)
+ }
+ for i, m := range want {
+ if recs[i].Msg != m {
+ t.Fatalf("order[%d] = %q, want %q", i, recs[i].Msg, m)
+ }
+ }
+
+ // Stream filter composes.
+ recs, _, err = ReadWindow(ctx, s, "bde", "lpay", Filter{Stream: StreamPlatform}, 1000)
+ if err != nil || len(recs) != 1 || recs[0].Level != LevelError {
+ t.Fatalf("platform filter: %v %+v", err, recs)
+ }
+
+ // Unbounded window sees all three hours.
+ recs, _, err = ReadWindow(ctx, s, "bde", "lpay", Filter{}, 1000)
+ if err != nil || len(recs) != 4 {
+ t.Fatalf("unbounded: %v, %d records", err, len(recs))
+ }
+
+ // collectCap keeps the TAIL and says it trimmed.
+ recs, trimmed, err = ReadWindow(ctx, s, "bde", "lpay", Filter{}, 2)
+ if err != nil || !trimmed || len(recs) != 2 {
+ t.Fatalf("capped: %v trimmed=%v n=%d", err, trimmed, len(recs))
+ }
+ if recs[len(recs)-1].Msg != "wake failed: x" {
+ t.Fatalf("cap dropped the tail: %+v", recs)
+ }
+
+ // An app with no logs at all is an empty answer, not an error.
+ recs, _, err = ReadWindow(ctx, s, "bde", "ghost", Filter{}, 1000)
+ if err != nil || len(recs) != 0 {
+ t.Fatalf("empty app: %v %+v", err, recs)
+ }
+}
+
+func TestHourIntersects(t *testing.T) {
+ since := time.Date(2026, 8, 10, 11, 0, 0, 0, time.UTC)
+ for hour, want := range map[string]bool{
+ "2026-08-10-10": false, // ends exactly at since — nothing at/after since lives there
+ "2026-08-10-11": true,
+ "2026-08-10-12": true,
+ "not-an-hour": false,
+ } {
+ if got := hourIntersects(hour, since, time.Time{}); got != want {
+ t.Errorf("hourIntersects(%q, since) = %v, want %v", hour, got, want)
+ }
+ }
+ until := time.Date(2026, 8, 10, 11, 30, 0, 0, time.UTC)
+ if hourIntersects("2026-08-10-12", time.Time{}, until) {
+ t.Error("an hour starting after until intersects")
+ }
+ if !hourIntersects("2026-08-10-11", since, until) {
+ t.Error("the window's own hour should intersect")
+ }
+}
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..309b46d8a
--- /dev/null
+++ b/internal/applog/sink.go
@@ -0,0 +1,195 @@
+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 // monotonic across apps AND restarts — the follow cursor's spine
+
+ notify chan struct{} // poked (non-blocking) on Event; the ship loop's fast-flush signal
+}
+
+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). The sequence counter starts at the
+// construction instant's UnixMilli rather than zero: chunk keys embed the
+// first record's seq, and per-box seq order is a promise the cursor relies
+// on — a restart that re-based to zero would write keys that sort BEFORE
+// last boot's and break exact per-box ordering. Milliseconds always outrun
+// the admitted line rate (the rate cap is ~17 KiB/s), so a rebase lands
+// above everything the previous process numbered.
+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),
+ seq: uint64(now().UnixMilli()),
+ notify: make(chan struct{}, 1),
+ }
+}
+
+// Wake is the ship loop's fast-flush signal: it receives after an Event has
+// been buffered, so a wake failure reaches the bucket in seconds rather
+// than at the next tick (spec §5). Level-triggered and lossy by design —
+// one pending poke summarizes any number of events.
+func (s *Sink) Wake() <-chan struct{} { return s.notify }
+
+// Known lists every app this sink has ever buffered for, in no particular
+// order — the ship loop's retention worklist.
+func (s *Sink) Known() []AppKey {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ out := make([]AppKey, 0, len(s.apps))
+ for k := range s.apps {
+ out = append(out, k)
+ }
+ return out
+}
+
+// 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()
+ s.append(s.buf(AppKey{acct, app}), Record{
+ Host: host,
+ Stream: StreamPlatform,
+ Level: level,
+ Msg: msg,
+ })
+ s.mu.Unlock()
+ select { // non-blocking: a pending poke already says everything
+ case s.notify <- struct{}{}:
+ default:
+ }
+}
+
+// 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)
+ }
+}
diff --git a/internal/console/apilogs.go b/internal/console/apilogs.go
new file mode 100644
index 000000000..ae7d177e5
--- /dev/null
+++ b/internal/console/apilogs.go
@@ -0,0 +1,153 @@
+// The CLI's logs API — the read side of `carlos logs` (spec
+// 2026-08-10-carlos-logs-design.md §6): one owner-gated GET that merges the
+// app's shipped chunks across boxes, filters, and pages. It shares the
+// instance family's gate (instanceGate) deliberately: same principal proof,
+// same byte-identical 404, same rate limiter — and an APP token may read
+// its own app's logs, which is the same trust the instance verbs already
+// extend it.
+package console
+
+import (
+ "context"
+ "log"
+ "net/http"
+ "regexp"
+ "strconv"
+ "time"
+
+ "github.com/carlosframework/platform/internal/applog"
+)
+
+const (
+ // logsDefaultLimit / logsMaxLimit bound one answer (spec §6). The
+ // response's `truncated` says when the window held more.
+ logsDefaultLimit = 1000
+ logsMaxLimit = 5000
+ // logsCollectCap bounds what one read will hold in memory while
+ // assembling the window — the tail survives, and `truncated` reports it.
+ logsCollectCap = 50000
+ // logsDefaultWindow applies when the caller sends no `since`: an
+ // unbounded read would walk every hour of retention for no stated
+ // reason; a caller that wants more says so.
+ logsDefaultWindow = time.Hour
+)
+
+// logsResponse is the wire shape both the CLI and the console tab read.
+type logsResponse struct {
+ Records []applog.Record `json:"records"`
+ Truncated bool `json:"truncated"`
+ Cursor string `json:"cursor,omitempty"`
+}
+
+// handleAPILogs answers `carlos logs`: GET with since/until/stream/host/
+// level/grep/field/limit/cursor query params. Without a cursor the answer
+// is the TAIL of the window (the last `limit` records — what a log reader
+// came for); with one it is the records strictly after that position, which
+// is what makes `-f` a plain 2s poll of this same endpoint.
+func (s *Server) handleAPILogs(w http.ResponseWriter, r *http.Request) {
+ _, acct, app, ok := s.instanceGate(w, r)
+ if !ok {
+ return
+ }
+ f, limit, badParam := logsFilterFromQuery(r, time.Now())
+ if badParam != "" {
+ jsonError(w, http.StatusBadRequest, badParam)
+ return
+ }
+ ctx, cancel := context.WithTimeout(r.Context(), instanceStoreTimeout)
+ defer cancel()
+ recs, trimmed, err := applog.ReadWindow(ctx, s.cfg.Releases, acct.Sqid, app, f, logsCollectCap)
+ if err != nil {
+ log.Printf("console: api logs %s/%s: %v", acct.Sqid, app, err)
+ jsonError(w, http.StatusBadGateway, "could not read the logs")
+ return
+ }
+
+ resp := logsResponse{Truncated: trimmed}
+ if cur, ok := applog.DecodeCursor(r.URL.Query().Get("cursor")); ok {
+ // Follow: everything strictly after the cursor, oldest first.
+ var after []applog.Record
+ for _, rec := range recs {
+ if cur.Before(rec) {
+ after = append(after, rec)
+ }
+ }
+ if len(after) > limit {
+ after = after[:limit]
+ resp.Truncated = true
+ }
+ resp.Records = after
+ } else {
+ // Tail: the last `limit` records of the window.
+ if len(recs) > limit {
+ recs = recs[len(recs)-limit:]
+ resp.Truncated = true
+ }
+ resp.Records = recs
+ }
+ if n := len(resp.Records); n > 0 {
+ resp.Cursor = applog.EncodeCursor(resp.Records[n-1])
+ } else if c := r.URL.Query().Get("cursor"); c != "" {
+ resp.Cursor = c // nothing new — the caller keeps polling from here
+ }
+ if resp.Records == nil {
+ resp.Records = []applog.Record{} // "no logs" is [], not null
+ }
+ writeJSON(w, http.StatusOK, resp)
+}
+
+// logsFilterFromQuery builds the applog.Filter (plus record limit) from the
+// request. The empty badParam means the params parsed; anything else is the
+// 400 text, naming the parameter in the member's own terms.
+func logsFilterFromQuery(r *http.Request, now time.Time) (f applog.Filter, limit int, badParam string) {
+ q := r.URL.Query()
+ var err error
+ if f.Since, err = applog.ParseTimeArg(q.Get("since"), now); err != nil {
+ return f, 0, "since: " + err.Error()
+ }
+ if f.Until, err = applog.ParseTimeArg(q.Get("until"), now); err != nil {
+ return f, 0, "until: " + err.Error()
+ }
+ if f.Since.IsZero() && q.Get("cursor") == "" {
+ f.Since = now.Add(-logsDefaultWindow)
+ }
+ if v := q.Get("stream"); v != "" {
+ if !applog.ValidStream(v) {
+ return f, 0, "stream: want app, platform, edge or release"
+ }
+ f.Stream = v
+ }
+ f.Host = q.Get("host")
+ if v := q.Get("level"); v != "" {
+ if !applog.ValidLevel(v) {
+ return f, 0, "level: want debug, info, warn or error"
+ }
+ f.Level = v
+ }
+ if v := q.Get("grep"); v != "" {
+ re, err := regexp.Compile(v)
+ if err != nil {
+ return f, 0, "grep: " + err.Error()
+ }
+ f.Grep = re
+ }
+ for _, kv := range q["field"] {
+ k, v, err := applog.ParseFieldArg(kv)
+ if err != nil {
+ return f, 0, err.Error()
+ }
+ if f.Fields == nil {
+ f.Fields = map[string]string{}
+ }
+ f.Fields[k] = v
+ }
+ limit = logsDefaultLimit
+ if v := q.Get("limit"); v != "" {
+ n, err := strconv.Atoi(v)
+ if err != nil || n < 1 {
+ return f, 0, "limit: want a positive integer"
+ }
+ limit = min(n, logsMaxLimit)
+ }
+ return f, limit, ""
+}
diff --git a/internal/console/apilogs_test.go b/internal/console/apilogs_test.go
new file mode 100644
index 000000000..46ca0b002
--- /dev/null
+++ b/internal/console/apilogs_test.go
@@ -0,0 +1,183 @@
+package console
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/applog"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func logsURL(fx *consoleFixture, acct, app, query string) string {
+ u := fx.ts.URL + "/api/cli/apps/" + acct + "/" + app + "/logs"
+ if query != "" {
+ u += "?" + query
+ }
+ return u
+}
+
+// seedLogChunks ships a small timeline for titogo: an app line and a wake
+// failure, recent enough for the default 1h window.
+func seedLogChunks(t *testing.T, fx *consoleFixture, now time.Time) []applog.Record {
+ t.Helper()
+ recs := []applog.Record{
+ {TS: now.Add(-10 * time.Minute), Box: "box1", Host: "titogo.example.com",
+ Stream: applog.StreamApp, Msg: "listening on socket", Seq: 100,
+ Fields: map[string]string{"request_id": "r1"}},
+ {TS: now.Add(-5 * time.Minute), Box: "box1", Host: "titogo.example.com",
+ Stream: applog.StreamPlatform, Level: applog.LevelError,
+ Msg: "wake failed: nothing is promoted for this instance's channel", Seq: 101},
+ }
+ body, err := applog.EncodeChunk(recs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ key := applog.ChunkKey(release.OpsAccount, "titogo", "box1", applog.HourOf(recs[0].TS), recs[0].Seq, len(body))
+ if _, err := fx.rel.Put(context.Background(), key, body, ""); err != nil {
+ t.Fatal(err)
+ }
+ return recs
+}
+
+// TestAPILogsTailFiltersAndFollow drives the endpoint end to end: default
+// tail, stream/level/grep/field filters, and the cursor follow loop.
+func TestAPILogsTailFiltersAndFollow(t *testing.T) {
+ fx, token := instanceFixture(t, "code-logs-1")
+ now := time.Now().UTC()
+ seedLogChunks(t, fx, now)
+
+ get := func(query string) logsResponse {
+ t.Helper()
+ status, _, body := rawReq(t, http.MethodGet, logsURL(fx, fx.sqid, "titogo", query), token, "")
+ if status != http.StatusOK {
+ t.Fatalf("GET ?%s = %d: %s", query, status, body)
+ }
+ var resp logsResponse
+ if err := json.Unmarshal([]byte(body), &resp); err != nil {
+ t.Fatalf("bad JSON: %v: %s", err, body)
+ }
+ return resp
+ }
+
+ // Default window: both records, oldest first, cursor present.
+ resp := get("")
+ if len(resp.Records) != 2 || resp.Records[0].Msg != "listening on socket" {
+ t.Fatalf("tail: %+v", resp)
+ }
+ if resp.Cursor == "" || resp.Truncated {
+ t.Fatalf("cursor/truncated: %+v", resp)
+ }
+
+ // Filters compose: stream, level, grep, field.
+ if r := get("stream=platform"); len(r.Records) != 1 || r.Records[0].Level != "error" {
+ t.Fatalf("stream filter: %+v", r.Records)
+ }
+ if r := get("level=error"); len(r.Records) != 1 {
+ t.Fatalf("level filter: %+v", r.Records)
+ }
+ if r := get("grep=promoted"); len(r.Records) != 1 || r.Records[0].Stream != "platform" {
+ t.Fatalf("grep filter: %+v", r.Records)
+ }
+ if r := get("field=request_id%3Dr1"); len(r.Records) != 1 || r.Records[0].Stream != "app" {
+ t.Fatalf("field filter: %+v", r.Records)
+ }
+ if r := get("host=titogo.example.com"); len(r.Records) != 2 {
+ t.Fatalf("host filter: %+v", r.Records)
+ }
+ if r := get("host=other.example.com"); len(r.Records) != 0 {
+ t.Fatalf("host mismatch filter: %+v", r.Records)
+ }
+
+ // limit=1 tails to the NEWEST record and says truncated.
+ r := get("limit=1")
+ if len(r.Records) != 1 || r.Records[0].Stream != "platform" || !r.Truncated {
+ t.Fatalf("limit tail: %+v", r)
+ }
+
+ // Follow: the cursor from the full read sees nothing new, keeps itself.
+ full := get("")
+ fr := get("cursor=" + full.Cursor)
+ if len(fr.Records) != 0 || fr.Cursor != full.Cursor {
+ t.Fatalf("follow at tip: %+v", fr)
+ }
+ // A new record lands; the same cursor now yields exactly it.
+ extra := []applog.Record{{TS: now.Add(-time.Minute), Box: "box1",
+ Host: "titogo.example.com", Stream: applog.StreamApp, Msg: "fresh line", Seq: 102}}
+ body, err := applog.EncodeChunk(extra)
+ if err != nil {
+ t.Fatal(err)
+ }
+ key := applog.ChunkKey(release.OpsAccount, "titogo", "box1", applog.HourOf(extra[0].TS), 102, len(body))
+ if _, err := fx.rel.Put(context.Background(), key, body, ""); err != nil {
+ t.Fatal(err)
+ }
+ fr2 := get("cursor=" + full.Cursor)
+ if len(fr2.Records) != 1 || fr2.Records[0].Msg != "fresh line" {
+ t.Fatalf("follow after new record: %+v", fr2)
+ }
+}
+
+// TestAPILogsRefusals: a non-member's read is the family's byte-identical
+// 404; bad parameters name themselves in a 400.
+func TestAPILogsRefusals(t *testing.T) {
+ fx, _ := instanceFixture(t, "code-logs-2")
+ eve := cliToken(t, fx, "eve@keymail.dev", "code-logs-eve")
+ want := canonicalNotFound(t)
+ status, h, body := rawReq(t, http.MethodGet, logsURL(fx, fx.sqid, "titogo", ""), eve, "")
+ assertNotFound(t, "logs as non-member", want, status, h, body)
+
+ _, token := func() (*consoleFixture, string) { return fx, cliToken(t, fx, "alice@keymail.dev", "code-logs-3") }()
+ for _, q := range []string{"since=garbage", "stream=nope", "level=shouting", "grep=%5B", "field=nokey", "limit=zero"} {
+ status, _, body := rawReq(t, http.MethodGet, logsURL(fx, fx.sqid, "titogo", q), token, "")
+ if status != http.StatusBadRequest {
+ t.Errorf("?%s = %d, want 400 (%s)", q, status, body)
+ }
+ }
+
+ // An app with no logs at all: an empty list, not an error and not null.
+ status, _, body = rawReq(t, http.MethodGet, logsURL(fx, fx.sqid, "titogo", ""), token, "")
+ if status != http.StatusOK || !json.Valid([]byte(body)) {
+ t.Fatalf("empty logs: %d %s", status, body)
+ }
+ var resp logsResponse
+ if err := json.Unmarshal([]byte(body), &resp); err != nil || resp.Records == nil {
+ t.Fatalf("records must be [], got %s", body)
+ }
+ _ = fmt.Sprintf("%v", resp)
+}
+
+// TestAppLogsPage: the console tab renders the same timeline — filters as
+// GET params, member-gated like every app page, honest empty state.
+func TestAppLogsPage(t *testing.T) {
+ fx, _ := instanceFixture(t, "code-logs-page")
+ now := time.Now().UTC()
+ seedLogChunks(t, fx, now)
+ fx.setUser("paul@keymail.dev")
+ c := newBrowser(t)
+ signUp(t, fx, c, "code-logs-signin")
+
+ status, _, body := get(t, c, fx.u("/apps/titogo/logs"))
+ if status != http.StatusOK {
+ t.Fatalf("logs page: %d %s", status, body)
+ }
+ for _, want := range []string{"listening on socket", "wake failed", "carlos logs -app titogo -f"} {
+ if !strings.Contains(body, want) {
+ t.Errorf("page missing %q", want)
+ }
+ }
+ // Stream filter narrows.
+ status, _, body = get(t, c, fx.u("/apps/titogo/logs?stream=platform"))
+ if status != http.StatusOK || strings.Contains(body, "listening on socket") || !strings.Contains(body, "wake failed") {
+ t.Fatalf("stream filter page: %d", status)
+ }
+ // Window with nothing in it: the honest empty state, still a 200.
+ status, _, body = get(t, c, fx.u("/apps/titogo/logs?since=1ns"))
+ if status != http.StatusOK || !strings.Contains(body, "No log lines in this window") {
+ t.Fatalf("empty window page: %d", status)
+ }
+}
diff --git a/internal/console/authcli.go b/internal/console/authcli.go
index 29b0facbb..168d53380 100644
--- a/internal/console/authcli.go
+++ b/internal/console/authcli.go
@@ -266,6 +266,9 @@ func (s *Server) registerCLIRoutes() {
// so a host named "instances-manifest" could still never reach it.
s.mux.HandleFunc("PUT /api/cli/apps/{account}/{app}/instances-manifest", s.handleAPIInstanceEnable)
s.mux.HandleFunc("GET /api/cli/apps/{account}/{app}/instances", s.handleAPIInstanceList)
+ // The logs read (`carlos logs`, apilogs.go) — same gate, same 404
+ // discipline, and an app token may read its own app's logs.
+ s.mux.HandleFunc("GET /api/cli/apps/{account}/{app}/logs", s.handleAPILogs)
s.mux.HandleFunc("POST /api/cli/apps/{account}/{app}/instances", s.handleAPIInstanceCreate)
s.mux.HandleFunc("GET /api/cli/apps/{account}/{app}/instances/{host}", s.handleAPIInstanceGet)
s.mux.HandleFunc("DELETE /api/cli/apps/{account}/{app}/instances/{host}", s.handleAPIInstanceDelete)
diff --git a/internal/console/server.go b/internal/console/server.go
index d249e3c39..931103b06 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -22,6 +22,7 @@ import (
"strings"
"time"
+ "github.com/carlosframework/platform/internal/applog"
"github.com/carlosframework/platform/internal/domains"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/release"
@@ -320,6 +321,7 @@ func NewServer(cfg Config, st *Store) (*Server, error) {
s.mux.HandleFunc("GET /a/{sqid}/apps/{app}", s.handleApp)
s.mux.HandleFunc("GET /a/{sqid}/apps/{app}/releases", s.handleAppReleases)
s.mux.HandleFunc("GET /a/{sqid}/apps/{app}/instances", s.handleAppInstances)
+ s.mux.HandleFunc("GET /a/{sqid}/apps/{app}/logs", s.handleAppLogs)
s.mux.HandleFunc("GET /a/{sqid}/apps/{app}/activity", s.handleAppActivity)
s.mux.HandleFunc("GET /a/{sqid}/apps/{app}/settings", s.handleAppSettings)
s.mux.HandleFunc("POST /a/{sqid}/apps/{app}/promote", s.handlePromote)
@@ -523,6 +525,15 @@ type page struct {
// App Instances: the one table — routes joined with their readings.
InstanceRows []InstanceRow
+ // App Logs (apilogs.go's sibling page): the merged timeline plus the
+ // filter state the form re-renders with. Empty LogRows with LogsErr ""
+ // is the normal nothing-logged-yet state, never an error.
+ LogRows []applog.Record
+ LogStream string
+ LogSince string
+ LogTruncated bool
+ LogsErr string
+
// Action confirm interstitial (confirm.go): what is about to happen,
// carried fields and all. Nil everywhere except the confirm page.
Confirm *ConfirmView
@@ -1603,6 +1614,49 @@ func (s *Server) handleAppInstances(w http.ResponseWriter, r *http.Request) {
s.render(w, http.StatusOK, "app_instances", p)
}
+// appLogsPageCap bounds the Logs tab to what a page can usefully show; the
+// CLI (`carlos logs`) is the door for more, and the page says so.
+const appLogsPageCap = 200
+
+// handleAppLogs renders the app's merged log timeline — the same window the
+// CLI reads (apilogs.go), tailed to the newest appLogsPageCap records, with
+// stream and window filters as plain GET params so a view is a URL.
+func (s *Server) handleAppLogs(w http.ResponseWriter, r *http.Request) {
+ p, acct, ok := s.appPage(w, r, "logs")
+ if !ok {
+ return
+ }
+ q := r.URL.Query()
+ p.LogStream = q.Get("stream")
+ if p.LogStream != "" && !applog.ValidStream(p.LogStream) {
+ p.LogStream = ""
+ }
+ p.LogSince = q.Get("since")
+ if _, err := time.ParseDuration(p.LogSince); err != nil {
+ p.LogSince = "1h"
+ }
+ f := applog.Filter{Stream: p.LogStream}
+ d, _ := time.ParseDuration(p.LogSince)
+ f.Since = time.Now().Add(-d)
+
+ if s.cfg.Releases != nil {
+ ctx, cancel := context.WithTimeout(r.Context(), instanceStoreTimeout)
+ defer cancel()
+ recs, trimmed, err := applog.ReadWindow(ctx, s.cfg.Releases, acct.Sqid, p.AppName, f, logsCollectCap)
+ if err != nil {
+ log.Printf("console: app logs %s/%s: %v", acct.Sqid, p.AppName, err)
+ p.LogsErr = "The logs could not be read just now — try again in a moment."
+ }
+ p.LogTruncated = trimmed
+ if len(recs) > appLogsPageCap {
+ recs = recs[len(recs)-appLogsPageCap:]
+ p.LogTruncated = true
+ }
+ p.LogRows = recs
+ }
+ s.render(w, http.StatusOK, "app_logs", p)
+}
+
// handleAppActivity is the console's own audit trail for this app.
func (s *Server) handleAppActivity(w http.ResponseWriter, r *http.Request) {
p, acct, ok := s.appPage(w, r, "activity")
diff --git a/internal/console/templates/app_logs.html b/internal/console/templates/app_logs.html
new file mode 100644
index 000000000..e04d66787
--- /dev/null
+++ b/internal/console/templates/app_logs.html
@@ -0,0 +1,50 @@
+{{define "app_logs"}}{{template "top" .}}
+{{if .Flash}}<p class="banner">{{.Flash}}</p>{{end}}
+{{if .FlashErr}}<p class="banner error">{{.FlashErr}}</p>{{end}}
+{{template "appnav" .}}
+
+<h2>Logs</h2>
+{{if .LogsErr}}<p class="banner error">{{.LogsErr}}</p>{{end}}
+<form method="get" class="small">
+ <label>Show
+ <select name="stream">
+ <option value=""{{if eq .LogStream ""}} selected{{end}}>everything</option>
+ <option value="app"{{if eq .LogStream "app"}} selected{{end}}>app output</option>
+ <option value="platform"{{if eq .LogStream "platform"}} selected{{end}}>platform events</option>
+ </select>
+ </label>
+ <label>from the last
+ <select name="since">
+ <option value="15m"{{if eq .LogSince "15m"}} selected{{end}}>15 minutes</option>
+ <option value="1h"{{if eq .LogSince "1h"}} selected{{end}}>hour</option>
+ <option value="6h"{{if eq .LogSince "6h"}} selected{{end}}>6 hours</option>
+ <option value="24h"{{if eq .LogSince "24h"}} selected{{end}}>day</option>
+ <option value="168h"{{if eq .LogSince "168h"}} selected{{end}}>week</option>
+ </select>
+ </label>
+ <button type="submit">Update</button>
+</form>
+
+{{if .LogRows}}
+{{if .LogTruncated}}<p class="small">Showing the newest {{len .LogRows}} lines of this window — narrow the window, or use <code>carlos logs</code> for more.</p>{{end}}
+<div class="tablewrap">
+<table class="logs">
+ <tr><th scope="col">When</th><th scope="col">Stream</th><th scope="col">Instance</th><th scope="col">Message</th></tr>
+ {{range .LogRows}}
+ <tr>
+ <td class="small">{{fmtTime .TS}}</td>
+ <td><code>{{.Stream}}</code></td>
+ <td class="small">{{.Host}}</td>
+ <td>{{if eq .Level "error"}}<strong>{{.Msg}}</strong>{{else}}{{.Msg}}{{end}}</td>
+ </tr>
+ {{end}}
+</table>
+</div>
+<p class="small">Times are UTC. Follow live with <code>carlos logs -app {{.AppName}} -f</code>.</p>
+{{else if not .LogsErr}}
+<p>No log lines in this window.</p>
+<p class="small">An app's output appears here shortly after it runs; platform events
+(wakes, restarts, failures) appear even when the app itself printed nothing. A
+site served from files alone has no app output.</p>
+{{end}}
+{{template "bottom" .}}{{end}}
diff --git a/internal/console/templates/appnav.html b/internal/console/templates/appnav.html
index d6696818c..c4b4a3a11 100644
--- a/internal/console/templates/appnav.html
+++ b/internal/console/templates/appnav.html
@@ -7,6 +7,7 @@
<a href="{{.Nav}}/apps/{{.AppName}}"{{if eq .ActiveTab "overview"}} aria-current="page"{{end}}>Overview</a>
<a href="{{.Nav}}/apps/{{.AppName}}/releases"{{if eq .ActiveTab "releases"}} aria-current="page"{{end}}>Versions</a>
<a href="{{.Nav}}/apps/{{.AppName}}/instances"{{if eq .ActiveTab "instances"}} aria-current="page"{{end}}>Running</a>
+ <a href="{{.Nav}}/apps/{{.AppName}}/logs"{{if eq .ActiveTab "logs"}} aria-current="page"{{end}}>Logs</a>
<a href="{{.Nav}}/apps/{{.AppName}}/activity"{{if eq .ActiveTab "activity"}} aria-current="page"{{end}}>Activity</a>
<a href="{{.Nav}}/apps/{{.AppName}}/settings"{{if eq .ActiveTab "settings"}} aria-current="page"{{end}}>Settings</a>
</nav>
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 94a784712..c9051bec2 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -91,6 +91,7 @@ import (
"github.com/carlosframework/platform/internal/activator"
"github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/applog"
"github.com/carlosframework/platform/internal/caps"
"github.com/carlosframework/platform/internal/cgroup"
"github.com/carlosframework/platform/internal/domains"
@@ -139,6 +140,15 @@ func RunAgent(args []string) {
// per-instance readings.
cgm := cgroup.Setup(log.Printf)
+ // Log shipping (`carlos logs`): one sink for every producer in this
+ // process, armed only when a bucket exists to ship to — a nil sink
+ // leaves every producer exactly as before (no tees, no pipes, no
+ // events). The box identity is the one metrics publishes under.
+ var logSink *applog.Sink
+ if resolveStatusBucket(os.Getenv) != "" {
+ logSink = applog.NewSink(status.Label(readMachineID()), time.Now)
+ }
+
cfg, err := activator.ConfigFromEnv()
if err != nil {
log.Fatalf("agent: %v", err)
@@ -150,17 +160,24 @@ func RunAgent(args []string) {
// the split backend dispatches per route on Backing. Note the arming
// gate is still the hibernate bucket env even though unit tenants
// never touch the bucket — one opt-in switch for the whole activator.
+ eb := &activator.ExecBackend{
+ Cfg: *cfg,
+ OptDir: release.OptDir(),
+ DataDir: dd,
+ Cache: cache,
+ Materialize: materializeHook(reg, dd),
+ Cgroups: cgm,
+ }
+ if logSink != nil { // a typed-nil in the interface would NOT be nil
+ eb.Logs = logSink
+ }
svc = activator.New(*cfg, &activator.SplitBackend{
- Exec: &activator.ExecBackend{
- Cfg: *cfg,
- OptDir: release.OptDir(),
- DataDir: dd,
- Cache: cache,
- Materialize: materializeHook(reg, dd),
- Cgroups: cgm,
- },
+ Exec: eb,
Unit: &activator.UnitBackend{OptDir: release.OptDir()},
}, reg) // the registry satisfies activator.DueStore directly
+ if logSink != nil {
+ svc.Logs = logSink
+ }
svc.Start()
act = svc
log.Printf("agent: hibernation activator armed (bucket %s, idle %s)", cfg.Bucket, cfg.Idle)
@@ -168,6 +185,9 @@ func RunAgent(args []string) {
go adoptLoop(reg, svc)
go fastConvergeLoop(reg, svc)
+ if logSink != nil {
+ go logshipLoop(logSink, status.Label(readMachineID()))
+ }
go statusLoop(reg)
go domainsLoop(reg, peekAcmeCache(args))
go instancesLoop(reg)
diff --git a/internal/host/logship.go b/internal/host/logship.go
new file mode 100644
index 000000000..6861c0a28
--- /dev/null
+++ b/internal/host/logship.go
@@ -0,0 +1,214 @@
+// The log shipping loop — the box-side pump of `carlos logs` (spec
+// 2026-08-10-carlos-logs-design.md §5). Producers (the activator's exec
+// tees and platform events) buffer into one applog.Sink; every 10s — or
+// within ~2s of a platform event, which is what lets `carlos deploy`'s
+// timeout window see the wake error — this loop drains it and writes one
+// immutable gzipped chunk per app into the deployment bucket. On each hour
+// boundary it prunes ITS OWN box subtree per app: hours beyond the age
+// floor, then oldest-first beyond the byte cap (sizes ride in the chunk
+// keys — applog.ChunkKeySize — so no Store interface change). The
+// sampler/flush split mirrors metricsLoop for the same reason: the loop
+// shares a process with the box's traffic, so it recovers panics and
+// disables itself without a bucket rather than taking the box down.
+package host
+
+import (
+ "context"
+ "log"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/applog"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const (
+ logshipTick = 10 * time.Second
+ logshipDebounce = 2 * time.Second // event-poke settle: batch a wake's burst into one PUT
+ logshipMaxAge = 7 * 24 * time.Hour
+ logshipMaxBytes = 50 << 20 // per app per box
+ // logshipMaxPending caps records held across failed flushes, per app.
+ // The bucket being unwritable must not grow the agent's heap forever
+ // (samplerMaxBuf's reasoning); oldest go first, silently — a box-wide
+ // bucket outage is loudly visible elsewhere, and the gap in seq numbers
+ // records that something is missing.
+ logshipMaxPending = 4096
+)
+
+func logshipLoop(sink *applog.Sink, box string) {
+ // Same rule as metricsLoop: this loop shares a process with the agent
+ // serving traffic — log shipping stopping is a bad day, the box
+ // stopping is an outage.
+ defer func() {
+ if r := recover(); r != nil {
+ log.Printf("agent: logship loop panic, log shipping stopped for this process: %v", r)
+ }
+ }()
+
+ ctx := context.Background()
+ bucket := resolveStatusBucket(os.Getenv)
+ if bucket == "" {
+ log.Printf("agent: logship loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set")
+ return
+ }
+ s, err := release.NewS3Store(ctx, bucket)
+ if err != nil {
+ log.Printf("agent: logship loop disabled: %v", err)
+ return
+ }
+ log.Printf("agent: logship loop armed (10s ticks, event fast-flush), publishing to bucket %s", bucket)
+
+ pending := make(map[applog.AppKey][]applog.Record)
+ pruned := "" // last hour a prune ran for, "" = not yet
+ ticker := time.NewTicker(logshipTick)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ticker.C:
+ case <-sink.Wake():
+ time.Sleep(logshipDebounce) // let the event's burst finish
+ }
+ logshipGather(sink, pending)
+ logshipFlush(ctx, s, pending)
+
+ if hour := applog.HourOf(time.Now()); hour != pruned {
+ // Prune covers every app this box is CURRENTLY holding or has
+ // just flushed; an app that stops logging entirely ages out via
+ // other boxes' passes or the app's own deletion sweep. Sub-cap
+ // apps pay one ListDirs each per hour.
+ for _, k := range logshipKnownApps(pending, sink) {
+ logshipPrune(ctx, s, box, k, time.Now())
+ }
+ pruned = hour
+ }
+ }
+}
+
+// logshipGather drains the sink into pending, enforcing the per-app cap —
+// oldest first, so what survives a long bucket outage is the freshest tail.
+func logshipGather(sink *applog.Sink, pending map[applog.AppKey][]applog.Record) {
+ for k, recs := range sink.Drain() {
+ merged := append(pending[k], recs...)
+ if over := len(merged) - logshipMaxPending; over > 0 {
+ merged = merged[over:]
+ }
+ pending[k] = merged
+ }
+}
+
+// logshipFlush writes one chunk per pending app — create-only, keyed by the
+// batch's first seq, so a retried batch can never overwrite a delivered one
+// — and forgets what landed. A failed app keeps its records for the next
+// round; other apps' flushes proceed regardless.
+func logshipFlush(ctx context.Context, s release.Store, pending map[applog.AppKey][]applog.Record) {
+ for k, recs := range pending {
+ if len(recs) == 0 {
+ delete(pending, k)
+ continue
+ }
+ body, err := applog.EncodeChunk(recs)
+ if err != nil { // marshal of our own records — a bug, not a retry case
+ log.Printf("agent: logship encode %s/%s: %v (dropping %d records)", k.Acct, k.App, err, len(recs))
+ delete(pending, k)
+ continue
+ }
+ key := applog.ChunkKey(k.Acct, k.App, recs[0].Box, applog.HourOf(recs[0].TS), recs[0].Seq, len(body))
+ if _, err := s.Put(ctx, key, body, ""); err != nil {
+ log.Printf("agent: logship flush %s/%s: %v (%d records held)", k.Acct, k.App, err, len(recs))
+ continue
+ }
+ delete(pending, k)
+ }
+}
+
+// logshipKnownApps is the union of apps with pending records and apps the
+// sink has ever buffered for — the prune worklist.
+func logshipKnownApps(pending map[applog.AppKey][]applog.Record, sink *applog.Sink) []applog.AppKey {
+ seen := make(map[applog.AppKey]bool)
+ var out []applog.AppKey
+ for k := range pending {
+ if !seen[k] {
+ seen[k] = true
+ out = append(out, k)
+ }
+ }
+ for _, k := range sink.Known() {
+ if !seen[k] {
+ seen[k] = true
+ out = append(out, k)
+ }
+ }
+ return out
+}
+
+// logshipPrune enforces retention for ONE app on THIS box's subtree only —
+// each box owns what it wrote (spec §5). Hour granularity: an hour goes
+// when it is older than the age floor, or (oldest first) while the
+// subtree's key-embedded sizes total over the byte cap. Errors are logged
+// and left for the next hourly pass — retention is idempotent.
+func logshipPrune(ctx context.Context, s release.Store, box string, k applog.AppKey, now time.Time) {
+ prefix := applog.BoxPrefix(k.Acct, k.App, box)
+ hourDirs, err := s.ListDirs(ctx, prefix)
+ if err != nil {
+ log.Printf("agent: logship prune %s: %v", prefix, err)
+ return
+ }
+ // ListDirs returns full prefixes ending in "/", sorted is not promised —
+ // the hour names sort chronologically, so order them ourselves via the
+ // keys' lexical shape. Collect per-hour keys+sizes as we go.
+ type hourEnt struct {
+ dir string
+ keys []string
+ bytes int64
+ }
+ var hours []hourEnt
+ var total int64
+ for _, dir := range hourDirs {
+ keys, err := s.List(ctx, dir)
+ if err != nil {
+ log.Printf("agent: logship prune %s: %v", dir, err)
+ return
+ }
+ var b int64
+ for _, key := range keys {
+ if n, ok := applog.ChunkKeySize(key); ok {
+ b += n
+ }
+ }
+ hours = append(hours, hourEnt{dir: dir, keys: keys, bytes: b})
+ total += b
+ }
+ // hourDirs from both implementations come lexically sorted (S3 lists in
+ // key order; DirStore sorts), and hour names are zero-padded UTC — so
+ // hours[0] is oldest.
+ cutoff := now.Add(-logshipMaxAge)
+ for _, h := range hours {
+ hourName := hourNameOf(h.dir)
+ drop := applog.HourBefore(hourName, cutoff) || total > logshipMaxBytes
+ if !drop {
+ break // hours are oldest-first; nothing further is droppable
+ }
+ for _, key := range h.keys {
+ if err := s.Delete(ctx, key); err != nil {
+ log.Printf("agent: logship prune %s: %v", key, err)
+ return
+ }
+ }
+ total -= h.bytes
+ }
+}
+
+// hourNameOf extracts the hour directory name from its full prefix
+// (".../logs/<box>/<hour>/" → "<hour>").
+func hourNameOf(dir string) string {
+ trimmed := dir
+ if n := len(trimmed); n > 0 && trimmed[n-1] == '/' {
+ trimmed = trimmed[:n-1]
+ }
+ for i := len(trimmed) - 1; i >= 0; i-- {
+ if trimmed[i] == '/' {
+ return trimmed[i+1:]
+ }
+ }
+ return trimmed
+}
diff --git a/internal/host/logship_test.go b/internal/host/logship_test.go
new file mode 100644
index 000000000..38e8419b0
--- /dev/null
+++ b/internal/host/logship_test.go
@@ -0,0 +1,142 @@
+package host
+
+import (
+ "context"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/applog"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func logshipFixture(t *testing.T) (release.Store, *applog.Sink) {
+ t.Helper()
+ s := release.NewDirStore(t.TempDir())
+ now := time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)
+ return s, applog.NewSink("box1", func() time.Time { return now })
+}
+
+// TestLogshipFlushWritesOneChunkPerApp: a drain with two apps lands as two
+// immutable chunks, each under its own app prefix, decodable and complete.
+func TestLogshipFlushWritesOneChunkPerApp(t *testing.T) {
+ s, sink := logshipFixture(t)
+ sink.App("bde", "lpay", "lpay.bde.oncarlos.com", "hello")
+ sink.Event("bab", "kass", "kass.example.com", applog.LevelInfo, "instance woke")
+
+ pending := make(map[applog.AppKey][]applog.Record)
+ logshipGather(sink, pending)
+ logshipFlush(context.Background(), s, pending)
+ if len(pending) != 0 {
+ t.Fatalf("flushed apps still pending: %v", pending)
+ }
+
+ for _, app := range []applog.AppKey{{Acct: "bde", App: "lpay"}, {Acct: "bab", App: "kass"}} {
+ keys, err := s.List(context.Background(), applog.Prefix(app.Acct, app.App))
+ if err != nil || len(keys) != 1 {
+ t.Fatalf("chunks for %v: %v (%v)", app, keys, err)
+ }
+ body, _, err := s.Get(context.Background(), keys[0])
+ if err != nil {
+ t.Fatal(err)
+ }
+ recs, err := applog.DecodeChunk(body)
+ if err != nil || len(recs) != 1 {
+ t.Fatalf("chunk decode for %v: %d recs, %v", app, len(recs), err)
+ }
+ // The key carries the real byte size — retention counts with it.
+ if n, ok := applog.ChunkKeySize(keys[0]); !ok || n != int64(len(body)) {
+ t.Fatalf("key size %d/%v, body %d: %s", n, ok, len(body), keys[0])
+ }
+ }
+}
+
+// TestLogshipFlushKeepsRecordsWhenTheBucketRefuses: a failed Put must hold
+// the app's records for the next round, not lose them — and the pending cap
+// bounds that holding.
+func TestLogshipFlushKeepsRecordsWhenTheBucketRefuses(t *testing.T) {
+ _, sink := logshipFixture(t)
+ sink.App("bde", "lpay", "h", "precious")
+ pending := make(map[applog.AppKey][]applog.Record)
+ logshipGather(sink, pending)
+ logshipFlush(context.Background(), failingStore{}, pending)
+ k := applog.AppKey{Acct: "bde", App: "lpay"}
+ if len(pending[k]) != 1 || pending[k][0].Msg != "precious" {
+ t.Fatalf("records lost on failed flush: %v", pending)
+ }
+ // Cap: a long outage trims oldest, keeps the freshest tail.
+ for i := 0; i < logshipMaxPending+10; i++ {
+ sink.App("bde", "lpay", "h", "line")
+ }
+ logshipGather(sink, pending)
+ if len(pending[k]) != logshipMaxPending {
+ t.Fatalf("pending cap not enforced: %d", len(pending[k]))
+ }
+ if pending[k][0].Msg == "precious" {
+ t.Fatal("cap kept oldest instead of freshest")
+ }
+}
+
+// TestLogshipPrune: hours past the age floor go; then oldest-first until
+// the key-embedded sizes fit the byte cap; the box only touches its own
+// subtree.
+func TestLogshipPrune(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+ now := time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)
+ k := applog.AppKey{Acct: "bde", App: "lpay"}
+
+ put := func(box, hour string, seq uint64, size int) string {
+ key := applog.ChunkKey(k.Acct, k.App, box, hour, seq, size)
+ if _, err := s.Put(ctx, key, make([]byte, size), ""); err != nil {
+ t.Fatal(err)
+ }
+ return key
+ }
+ oldKey := put("box1", "2026-08-01-10", 1, 10) // over the 7d floor
+ bigKey := put("box1", "2026-08-10-10", 2, 60<<20) // pushes total over the byte cap
+ liveKey := put("box1", "2026-08-10-11", 3, 10)
+ otherBox := put("box2", "2026-08-01-10", 4, 10) // NOT ours to prune
+
+ logshipPrune(ctx, s, "box1", k, now)
+
+ for key, want := range map[string]bool{oldKey: false, bigKey: false, liveKey: true, otherBox: true} {
+ _, _, err := s.Get(ctx, key)
+ if got := err == nil; got != want {
+ t.Errorf("%s survived=%v, want %v", key, got, want)
+ }
+ }
+}
+
+// failingStore refuses writes — the bucket-outage fixture.
+type failingStore struct{ release.Store }
+
+func (failingStore) Put(context.Context, string, []byte, string) (string, error) {
+ return "", context.DeadlineExceeded
+}
+
+// TestHourNameOf pins the prefix→name extraction.
+func TestHourNameOf(t *testing.T) {
+ if got := hourNameOf("apps/bde/lpay/logs/box1/2026-08-10-11/"); got != "2026-08-10-11" {
+ t.Fatalf("hourNameOf = %q", got)
+ }
+}
+
+// TestLogshipPruneKeepsRecentUnderCap: nothing under both limits is touched.
+func TestLogshipPruneKeepsRecentUnderCap(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+ now := time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)
+ k := applog.AppKey{Acct: "bde", App: "lpay"}
+ key := applog.ChunkKey(k.Acct, k.App, "box1", "2026-08-10-11", 1, 10)
+ if _, err := s.Put(ctx, key, make([]byte, 10), ""); err != nil {
+ t.Fatal(err)
+ }
+ logshipPrune(ctx, s, "box1", k, now)
+ if _, _, err := s.Get(ctx, key); err != nil {
+ t.Fatalf("live chunk pruned: %v", err)
+ }
+ if !strings.HasPrefix(key, "apps/bde/lpay/logs/") {
+ t.Fatalf("key layout drifted: %s", key)
+ }
+}
diff --git a/logs.go b/logs.go
new file mode 100644
index 000000000..558f7fdb0
--- /dev/null
+++ b/logs.go
@@ -0,0 +1,177 @@
+// `carlos logs` — an app's merged log timeline with no box access (spec
+// 2026-08-10-carlos-logs-design.md §7, issues #131/#78): the app's own
+// output plus the platform's events about it (wakes, restarts, failures),
+// read through the console API. API mode ONLY, deliberately: the journal on
+// a box was never the member surface, and the box-local form is exactly the
+// gap this verb closes.
+package main
+
+import (
+ "context"
+ "flag"
+ "fmt"
+ "io"
+ "log"
+ "net/url"
+ "os"
+ "sort"
+ "strconv"
+ "time"
+
+ "github.com/carlosframework/platform/internal/applog"
+)
+
+const logsUsage = `usage: carlos logs -app <a> [flags]
+
+Shows the newest lines first-to-last: the app's own stdout/stderr ("app")
+merged with the platform's events about it ("platform" — wakes, restarts,
+failures with the reason). A site served from files alone has no app
+stream; platform events still appear.
+
+flags:
+ -app <a> app whose logs to read (required)
+ -since <t> window start: a duration back from now (15m, 1h, 24h)
+ or an RFC3339 time (default 1h)
+ -until <t> window end, same forms (default: now)
+ -stream <s> only one stream: app | platform | edge | release
+ -host <h> only one instance (its route host)
+ -level <l> minimum level: debug | info | warn | error — only lines
+ that CARRY a level; plain text never claims one
+ -grep <re> only lines whose message matches this RE2 pattern
+ -field k=v only structured lines with this field (repeatable)
+ -n <n> at most this many lines (default 1000, max 5000)
+ -f follow: poll for new lines every 2s until interrupted
+ -account <a> account (sqid or name) the app belongs to (default "ops")
+ -account-id <id> account sqid, matched as a sqid ONLY — supersedes -account
+ -console <o> console to read through (default: the only logged-in one)
+`
+
+// logsFollowInterval is the -f poll cadence — the serial poll's own 2s, the
+// deployment's one convergence heartbeat.
+const logsFollowInterval = 2 * time.Second
+
+// logsMaxPolls bounds -f for the tests' benefit: 0 = forever (production).
+var logsMaxPolls = 0
+
+func runLogs(args []string) {
+ fs := flag.NewFlagSet("logs", flag.ExitOnError)
+ app := fs.String("app", "", "app whose logs to read")
+ since := fs.String("since", "1h", "window start: duration back from now, or RFC3339")
+ until := fs.String("until", "", "window end (default: now)")
+ stream := fs.String("stream", "", "only one stream: app | platform | edge | release")
+ host := fs.String("host", "", "only one instance (route host)")
+ level := fs.String("level", "", "minimum level: debug | info | warn | error")
+ grep := fs.String("grep", "", "only messages matching this RE2 pattern")
+ var fields multiFlag
+ fs.Var(&fields, "field", "only structured lines with this k=v field (repeatable)")
+ n := fs.Int("n", 0, "at most this many lines (default 1000, max 5000)")
+ follow := fs.Bool("f", false, "follow: poll for new lines until interrupted")
+ consoleFlag := fs.String("console", "", "console to read through (default: the only logged-in one)")
+ acctRaw := fs.String("account", "ops", "account (sqid or name) the app belongs to")
+ acctIDRaw := fs.String("account-id", "", "account sqid, matched as a sqid ONLY — supersedes -account")
+ fs.Usage = func() { fmt.Fprint(os.Stderr, logsUsage) }
+ fs.Parse(args)
+ if *app == "" {
+ log.Fatal(logsUsage)
+ }
+
+ q := url.Values{}
+ for k, v := range map[string]string{
+ "since": *since, "until": *until, "stream": *stream,
+ "host": *host, "level": *level, "grep": *grep,
+ } {
+ if v != "" {
+ q.Set(k, v)
+ }
+ }
+ for _, f := range fields {
+ q.Add("field", f)
+ }
+ if *n > 0 {
+ q.Set("limit", strconv.Itoa(*n))
+ }
+
+ sess := requireAPISession("logs", *consoleFlag)
+ acct := apiAccount(acctRaw, acctIDRaw)
+ ctx := context.Background()
+ res, err := sess.logsGet(ctx, acct, *app, q)
+ if err != nil {
+ log.Fatalf("logs: %v", err)
+ }
+ printLogRecords(stdout, res.Records)
+ if len(res.Records) == 0 && !*follow {
+ mutedf(stdout, "no log lines in this window — platform events appear on wakes and restarts; try -since 24h")
+ }
+ if res.Truncated {
+ mutedf(stdout, "(showing the newest %d lines of this window — narrow it, or raise -n)", len(res.Records))
+ }
+ if !*follow {
+ return
+ }
+
+ // Follow: the same endpoint, strictly-after the cursor, forever. The
+ // window params are dropped on purpose — a follow reads the PRESENT,
+ // and the server treats a cursor request as unbounded-forward.
+ fq := url.Values{}
+ for _, k := range []string{"stream", "host", "level", "grep"} {
+ if v := q.Get(k); v != "" {
+ fq.Set(k, v)
+ }
+ }
+ for _, f := range fields {
+ fq.Add("field", f)
+ }
+ cursor := res.Cursor
+ for polls := 0; logsMaxPolls == 0 || polls < logsMaxPolls; polls++ {
+ time.Sleep(logsFollowInterval)
+ if cursor != "" {
+ fq.Set("cursor", cursor)
+ }
+ res, err := sess.logsGet(ctx, acct, *app, fq)
+ if err != nil {
+ // A follow rides through a blip; the next poll answers or the
+ // member interrupts. Say it once per failure, quietly.
+ mutedf(stdout, "(logs: %v — still following)", err)
+ continue
+ }
+ printLogRecords(stdout, res.Records)
+ if res.Cursor != "" {
+ cursor = res.Cursor
+ }
+ }
+}
+
+// printLogRecords renders records one line each: UTC time, stream, host,
+// message (error lines red, warn yellow), then any structured fields dimmed.
+func printLogRecords(w io.Writer, recs []applog.Record) {
+ c := colorFor(w)
+ for _, r := range recs {
+ msg := r.Msg
+ switch r.Level {
+ case applog.LevelError:
+ msg = c.bad(msg)
+ case applog.LevelWarn:
+ msg = c.warn(msg)
+ }
+ line := c.muted(r.TS.UTC().Format("2006-01-02 15:04:05")) + " " +
+ c.head("["+r.Stream+"]") + " " + c.muted(r.Host) + " " + msg
+ keys := make([]string, 0, len(r.Fields))
+ for k := range r.Fields {
+ keys = append(keys, k)
+ }
+ sort.Strings(keys)
+ for _, k := range keys {
+ line += " " + c.muted(k+"="+r.Fields[k])
+ }
+ fmt.Fprintln(w, line)
+ }
+}
+
+// multiFlag collects a repeatable string flag.
+type multiFlag []string
+
+func (m *multiFlag) String() string { return fmt.Sprint([]string(*m)) }
+func (m *multiFlag) Set(v string) error {
+ *m = append(*m, v)
+ return nil
+}
diff --git a/logs_e2e_test.go b/logs_e2e_test.go
new file mode 100644
index 000000000..04ed54406
--- /dev/null
+++ b/logs_e2e_test.go
@@ -0,0 +1,159 @@
+// End-to-end acceptance for `carlos logs` (#131/#78): a REAL console.Server
+// over a DirStore holding shipped log chunks, a device-flow login, then the
+// real runLogs driven in-process — the member's whole path from "my app is
+// 503ing" to reading the wake failure, with no box access anywhere.
+// TestE2EInstancesThroughConsole's shape.
+package main
+
+import (
+ "context"
+ "net/http"
+ "net/http/httptest"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/applog"
+ "github.com/carlosframework/platform/internal/console"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func TestE2ELogsThroughConsole(t *testing.T) {
+ useTempCredentials(t)
+ clearDeploymentEnv(t)
+ checkNoRegistry := requireNoRegistryTouch(t)
+
+ tmp := t.TempDir()
+ relStore := release.NewDirStore(filepath.Join(tmp, "store"))
+ seed, pubB64, err := release.GenerateKey()
+ if err != nil {
+ t.Fatal(err)
+ }
+ priv, err := release.ParseSigningKey(seed, "e2e")
+ if err != nil {
+ t.Fatal(err)
+ }
+ pub, err := release.ParsePublicKey(pubB64, "e2e")
+ if err != nil {
+ t.Fatal(err)
+ }
+ cst, err := console.Open(filepath.Join(tmp, "console.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { cst.Close() })
+
+ var srv *console.Server
+ ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ srv.ServeHTTP(w, r)
+ }))
+ t.Cleanup(ts.Close)
+ srv, err = console.NewServer(console.Config{
+ Origin: ts.URL,
+ Keymail: "http://keymail.invalid", // never dialled — see e2eDeviceLogin
+ Registry: filepath.Join(tmp, "registry.db"),
+ Releases: relStore,
+ ReleasePub: pub,
+ Key: priv,
+ AppScheme: "http",
+ AppsDomain: "oncarlos.test",
+ }, cst)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ const addr = "paul@keymail.dev"
+ if err := cst.AddMember(console.OpsAccountID, addr, console.RoleOwner, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := cst.CreateAppIn(console.OpsAccountID, "titogo", addr, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ token, expiresAt := e2eDeviceLogin(t, ts.URL, cst, addr, "e2e@laptop")
+ credFile, err := loadCredentials()
+ if err != nil {
+ t.Fatal(err)
+ }
+ credFile.put(credential{Console: ts.URL, Address: addr, Token: token, ExpiresAt: expiresAt})
+ if err := credFile.save(); err != nil {
+ t.Fatal(err)
+ }
+
+ // A box shipped two chunks: routine output, then a wake failure — the
+ // #131 case, as the shipping loop would have written it.
+ now := time.Now().UTC()
+ writeChunk := func(seq uint64, recs []applog.Record) {
+ t.Helper()
+ body, err := applog.EncodeChunk(recs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ key := applog.ChunkKey(release.OpsAccount, "titogo", "box1", applog.HourOf(recs[0].TS), seq, len(body))
+ if _, err := relStore.Put(context.Background(), key, body, ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ writeChunk(100, []applog.Record{
+ {TS: now.Add(-20 * time.Minute), Box: "box1", Host: "titogo.bab.oncarlos.test",
+ Stream: applog.StreamApp, Msg: "listening on socket", Seq: 100,
+ Fields: map[string]string{"request_id": "r1"}},
+ })
+ writeChunk(101, []applog.Record{
+ {TS: now.Add(-2 * time.Minute), Box: "box1", Host: "titogo.bab.oncarlos.test",
+ Stream: applog.StreamPlatform, Level: applog.LevelError, Seq: 101,
+ Msg: "wake failed: nothing is promoted for this instance's channel"},
+ })
+
+ // The plain read: both streams merged, oldest first, wake reason visible.
+ buf := captureStdout(t)
+ runLogs([]string{"-app", "titogo", "-account", "ops"})
+ out := buf.String()
+ if !strings.Contains(out, "listening on socket") || !strings.Contains(out, "wake failed") {
+ t.Fatalf("logs stdout missing the timeline:\n%s", out)
+ }
+ if strings.Index(out, "listening on socket") > strings.Index(out, "wake failed") {
+ t.Fatalf("timeline out of order:\n%s", out)
+ }
+ if !strings.Contains(out, "[platform]") || !strings.Contains(out, "[app]") {
+ t.Fatalf("stream tags missing:\n%s", out)
+ }
+ if !strings.Contains(out, "request_id=r1") {
+ t.Fatalf("structured fields missing:\n%s", out)
+ }
+
+ // Filters reach the wire: -stream platform drops the app line.
+ buf.Reset()
+ runLogs([]string{"-app", "titogo", "-account", "ops", "-stream", "platform"})
+ if out := buf.String(); strings.Contains(out, "listening on socket") || !strings.Contains(out, "wake failed") {
+ t.Fatalf("-stream platform: %q", out)
+ }
+
+ // -f follows: a line shipped between polls arrives; the poll cap is the
+ // test's exit.
+ logsMaxPolls = 2
+ t.Cleanup(func() { logsMaxPolls = 0 })
+ done := make(chan string, 1)
+ go func() {
+ fbuf := captureStdout(t)
+ runLogs([]string{"-app", "titogo", "-account", "ops", "-f"})
+ done <- fbuf.String()
+ }()
+ time.Sleep(500 * time.Millisecond) // after the initial read, before poll 1 lands
+ writeChunk(102, []applog.Record{
+ {TS: now.Add(-time.Second), Box: "box1", Host: "titogo.bab.oncarlos.test",
+ Stream: applog.StreamApp, Msg: "fresh follow line", Seq: 102},
+ })
+ select {
+ case out := <-done:
+ if !strings.Contains(out, "fresh follow line") {
+ t.Fatalf("follow missed the new line:\n%s", out)
+ }
+ case <-time.After(15 * time.Second):
+ t.Fatal("follow did not finish its bounded polls")
+ }
+
+ // The whole flow ran through the console: nothing here may have touched
+ // a local registry.
+ checkNoRegistry()
+}
diff --git a/main.go b/main.go
index 58b38f113..62106f5e4 100644
--- a/main.go
+++ b/main.go
@@ -61,6 +61,8 @@ func main() {
runDeploy(args)
case "restart":
runRestart(args)
+ case "logs":
+ runLogs(args)
case "channels":
runChannels(args)
case "releases":
@@ -130,6 +132,8 @@ Your apps — ship and run software on a CARLOS deployment
env plain per-app config vars: set | unset | list
secrets sealed per-app secrets: set | unset | list | genkey
instances an app's instances: enable | create | list | delete
+ logs an app's merged log timeline: its own output plus platform
+ events (wakes, restarts, failures); -f follows live
ledger open ledgers: append | head | list | publish | blob | verify
vet check a release against the platform contract