feat(logs): System logs, log rules, and AI triage

config.example.yml

@@ -126,6 +126,9 @@ ai:
     api_key: your-gemini-api-key-here
     model: gemini-2.5-flash
     timeout: 1m0s
+    # Ceiling on log incidents the assistant may be asked to explain in a day.
+    # Unattended work is the only kind that can run up a bill unwatched.
+    triage_daily_cap: 25
 mcp:
     # Exposes the assistant's tool set to external MCP clients at /api/mcp. Every
     # call is authenticated and permission-gated exactly as the assistant is.

internal/api/internal_logs.go

@@ -0,0 +1,89 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+	"strconv"
+
+	"github.com/flatrun/agent/pkg/models"
+	"github.com/gin-gonic/gin"
+)
+
+// streamInternalLogs hands a deployment's log lines to a built-in app as newline-delimited
+// JSON. The user-facing stream is a websocket because a browser cannot set headers; an app
+// can, so it gets the simpler transport and the same reader, which keeps log sources, the
+// service filter and level parsing in one implementation.
+func (s *Server) streamInternalLogs(c *gin.Context) {
+	if s.pluginToken == "" || c.GetHeader("X-Plugin-Token") != s.pluginToken {
+		c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"})
+		return
+	}
+
+	name := c.Query("deployment")
+	if name == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "deployment required"})
+		return
+	}
+
+	deployment, err := s.manager.GetDeployment(name)
+	if err != nil {
+		c.JSON(http.StatusNotFound, gin.H{"error": "Deployment not found"})
+		return
+	}
+
+	source, ok := resolveLogSource(deployment.Metadata, c.Query("source"))
+	if !ok {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "unknown log source"})
+		return
+	}
+
+	services, err := s.resolveLogServices(name, c.Query("service"))
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+
+	// Replaying a day of history on every reconnect would re-raise handled incidents.
+	tail := 0
+	if v := c.Query("tail"); v != "" {
+		if n, parseErr := strconv.Atoi(v); parseErr == nil && n >= 0 {
+			tail = n
+		}
+	}
+
+	// Everything that can fail is resolved before the status goes out, since a 200 followed by
+	// silence is indistinguishable from a stream that has nothing to say yet.
+	var filePath string
+	if source.Type == models.LogSourceFile {
+		filePath, err = resolveLogFilePath(deployment.Path, source.Path)
+		if err != nil {
+			c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+			return
+		}
+	}
+
+	c.Writer.Header().Set("Content-Type", "application/x-ndjson")
+	c.Writer.Header().Set("Cache-Control", "no-cache")
+	c.Writer.WriteHeader(http.StatusOK)
+	c.Writer.Flush()
+
+	ctx := c.Request.Context()
+	encoder := json.NewEncoder(c.Writer)
+
+	sink := func(line string) {
+		record := parseLogRecord(line)
+		if source.Type == models.LogSourceFile && record.Service == "" {
+			record.Service = source.Name
+		}
+		if err := encoder.Encode(logLine{Type: "log", Line: line, Record: record}); err != nil {
+			return
+		}
+		c.Writer.Flush()
+	}
+
+	if source.Type == models.LogSourceFile {
+		_ = streamFileLogs(ctx, filePath, tail, sink)
+		return
+	}
+	_ = s.manager.StreamDeploymentLogs(ctx, name, deployment.Path, tail, sink, services...)
+}

internal/api/internal_logs_stream_test.go

@@ -0,0 +1,37 @@
+package api
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/flatrun/agent/internal/docker"
+	"github.com/gin-gonic/gin"
+)
+
+// A stream that answers 200 and then says nothing looks identical to one that has nothing to
+// report yet, so a source that cannot be read has to fail before the status goes out.
+func TestInternalLogStreamFailsBeforeAnsweringOK(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	base, name := writeLogFilterDeployment(t)
+
+	metadata := "log_sources:\n  - id: escape\n    name: Escape\n    type: file\n    path: ../outside.log\n"
+	if err := os.WriteFile(filepath.Join(base, name, "service.yml"), []byte(metadata), 0644); err != nil {
+		t.Fatal(err)
+	}
+
+	server := &Server{manager: docker.NewManager(base), pluginToken: "plugin-secret"}
+	router := gin.New()
+	router.GET("/internal/logs/stream", server.streamInternalLogs)
+
+	req := httptest.NewRequest(http.MethodGet, "/internal/logs/stream?deployment="+name+"&source=escape", nil)
+	req.Header.Set("X-Plugin-Token", "plugin-secret")
+	w := httptest.NewRecorder()
+	router.ServeHTTP(w, req)
+
+	if w.Code != http.StatusBadRequest {
+		t.Fatalf("expected 400, got %d: %s", w.Code, w.Body.String())
+	}
+}

internal/api/internal_logs_test.go

@@ -0,0 +1,53 @@
+package api
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+// The observability app reads this envelope to decide what is an incident. It is a wire
+// contract between two packages that are compiled together but talk over HTTP, so the shape
+// is pinned here and the app's watcher test decodes the same literal from the other side.
+func TestInternalLogEnvelopeShape(t *testing.T) {
+	raw := "web-1  | 2026-08-06T12:00:31.123456Z ERROR connection refused talking to redis"
+
+	encoded, err := json.Marshal(logLine{Type: "log", Line: raw, Record: parseLogRecord(raw)})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var decoded struct {
+		Type   string `json:"type"`
+		Line   string `json:"line"`
+		Record struct {
+			Timestamp string `json:"timestamp"`
+			Service   string `json:"service"`
+			Level     string `json:"level"`
+			Message   string `json:"message"`
+		} `json:"record"`
+	}
+	if err := json.Unmarshal(encoded, &decoded); err != nil {
+		t.Fatalf("the envelope must decode into the shape the watcher expects: %v", err)
+	}
+
+	if decoded.Type != "log" {
+		t.Errorf("type = %q, want log", decoded.Type)
+	}
+	if decoded.Line != raw {
+		t.Errorf("line should be the untouched original, got %q", decoded.Line)
+	}
+	if decoded.Record.Service != "web-1" {
+		t.Errorf("service should come from the compose prefix, got %q", decoded.Record.Service)
+	}
+	if decoded.Record.Level != "error" {
+		t.Errorf("level should be parsed to a canonical name, got %q", decoded.Record.Level)
+	}
+	// The compose prefix and the leading timestamp are stripped; the level word stays in the
+	// message, which is what the app fingerprints on.
+	if decoded.Record.Message != "ERROR connection refused talking to redis" {
+		t.Errorf("message should be the line without the compose prefix or timestamp, got %q", decoded.Record.Message)
+	}
+	if decoded.Record.Timestamp != "2026-08-06T12:00:31.123456Z" {
+		t.Errorf("timestamp should be lifted out of the line, got %q", decoded.Record.Timestamp)
+	}
+}