feat: harden runtime aggregation and matching

cmd/agentctl/main.go

@@ -424,7 +424,7 @@ func runRuntimeAudit(args []string, stdout, stderr io.Writer) error {
 		return fmt.Errorf("open runtime events: %w", err)
 	}
 	defer eventsFile.Close()
-	events, skipped, err := runtime.ReadSource(eventsFile, runtime.Source(*source), runtime.Options{})
+	runtimeReport, err := runtime.AggregateSource(eventsFile, runtime.Source(*source), runtime.Options{})
 	if err != nil {
 		return err
 	}
@@ -438,7 +438,7 @@ func runRuntimeAudit(args []string, stdout, stderr io.Writer) error {
 	if err := decoder.Decode(&inventory); err != nil {
 		return fmt.Errorf("decode inventory: %w", err)
 	}
-	audit := runtime.Audit(runtime.Aggregate(events, skipped), inventory)
+	audit := runtime.Audit(runtimeReport, inventory)
 	var payload []byte
 	if *format == "json" {
 		payload, err = json.MarshalIndent(audit, "", "  ")

docs/runtime-audit-mvp.md

@@ -36,7 +36,7 @@ MVP должен принимать нормализованные событи
 }
 ```
 
-Обязательные поля: `timestamp`, `agent_id` или `agent_name`, `operation`, `target`, `success`. Значения `operation` и `action` являются свободными строками на этапе MVP, чтобы не блокировать интеграции. Нормализатор должен отклонять пустые идентификаторы и некорректные timestamps.
+Обязательные поля: `timestamp`, `agent_id` или `agent_name`, `operation`, `target`, `success`. Формальная схема опубликована в [`docs/runtime-event.v1.schema.json`](./runtime-event.v1.schema.json). Значения `operation` и `action` являются свободными строками на этапе MVP, чтобы не блокировать интеграции. Нормализатор должен отклонять пустые идентификаторы и некорректные timestamps.
 
 ## Runtime-агрегат
 
@@ -53,7 +53,7 @@ MVP должен принимать нормализованные событи
 | `undeclared_targets` | Цели, отсутствующие в статическом inventory |
 | `undeclared_providers` | Провайдеры, отсутствующие в декларации |
 
-Агрегаты сортируются детерминированно. Это необходимо для стабильных diff-отчётов и baseline-механизма, уже используемого статическим сканером.
+Агрегаты сортируются детерминированно. Это необходимо для стабильных diff-отчётов и baseline-механизма, уже используемого статическим сканером. JSONL-источник агрегируется потоково и не удерживает весь список событий в памяти. Для envelope-форматов OpenTelemetry и API Gateway сначала выполняется безопасная нормализация, после чего применяется та же модель агрегата.
 
 ## Runtime findings
 
@@ -92,7 +92,7 @@ Runtime-аудит должен быть **наблюдателем, а не и
 
 Добавлена команда `agentctl runtime-audit <events> --inventory <report.json>`. Флаг `--source` выбирает `jsonl`, `otel-json` или `api-gateway`. Команда поддерживает text/json output, атомарную запись через существующий механизм `--output` и CI-порог `--fail-on`.
 
-Адаптеры принимают только metadata-поля. OpenTelemetry spans преобразуются по атрибутам агента, инструмента, окружения, provider и HTTP-статуса. API Gateway записи поддерживают snake_case и camelCase идентификаторы, JSONL и JSON-массив. Payload запроса и ответа не читается и не переносится в нормализованное событие.
+Адаптеры принимают только metadata-поля. OpenTelemetry spans преобразуются по атрибутам агента, инструмента, окружения, provider и HTTP-статуса. API Gateway записи поддерживают snake_case и camelCase идентификаторы, JSONL и JSON-массив. Payload запроса и ответа не читается и не переносится в нормализованное событие. Provider matching теперь выполняется на уровне конкретного агента: фактический provider считается заявленным только тогда, когда он принадлежит модели, указанной в `agent.models`.
 
 Следующий этап — добавить адаптеры OpenTelemetry и API gateway, не меняя нормализованный контракт событий.
 

docs/runtime-event.v1.schema.json

@@ -0,0 +1,24 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://github.com/HlinorAI/agent-control-plane/blob/main/docs/runtime-event.v1.schema.json",
+  "title": "Agent Control Plane runtime.v1 event",
+  "type": "object",
+  "additionalProperties": false,
+  "required": ["timestamp", "operation", "target", "success"],
+  "anyOf": [
+    {"required": ["agent_id"]},
+    {"required": ["agent_name"]}
+  ],
+  "properties": {
+    "timestamp": {"type": "string", "format": "date-time"},
+    "request_id": {"type": "string", "maxLength": 512},
+    "agent_id": {"type": "string", "minLength": 1, "maxLength": 512},
+    "agent_name": {"type": "string", "minLength": 1, "maxLength": 512},
+    "environment": {"type": "string", "maxLength": 128},
+    "operation": {"type": "string", "minLength": 1, "maxLength": 256},
+    "target": {"type": "string", "minLength": 1, "maxLength": 1024},
+    "provider": {"type": "string", "maxLength": 256},
+    "action": {"type": "string", "maxLength": 128},
+    "success": {"type": "boolean"}
+  }
+}

internal/runtime/adapters.go

@@ -34,6 +34,19 @@ func ReadSource(r io.Reader, source Source, options Options) ([]Event, int, erro
 	}
 }
 
+// AggregateSource streams JSONL directly into an aggregate report. Structured
+// adapter formats are normalized first because their envelope must be decoded.
+func AggregateSource(r io.Reader, source Source, options Options) (Report, error) {
+	if source == "" || source == SourceJSONL {
+		return AggregateReader(r, options)
+	}
+	events, skipped, err := ReadSource(r, source, options)
+	if err != nil {
+		return Report{}, err
+	}
+	return Aggregate(events, skipped), nil
+}
+
 type otelDocument struct {
 	ResourceSpans []otelResourceSpans `json:"resourceSpans"`
 }

internal/runtime/aggregate.go

@@ -6,51 +6,44 @@ import (
 )
 
 func Aggregate(events []Event, skipped int) Report {
-	byKey := make(map[string]*AgentSummary)
+	state := newAggregateState()
 	for _, event := range events {
-		key := event.AgentID
-		if key == "" {
-			key = "name:" + event.AgentName
-		}
-		summary := byKey[key]
-		if summary == nil {
-			summary = &AgentSummary{
-				AgentID:    event.AgentID,
-				AgentName:  event.AgentName,
-				FirstSeen:  event.Timestamp,
-				LastSeen:   event.Timestamp,
-				Operations: make(map[string]int),
-			}
-			byKey[key] = summary
-		}
-		summary.EventCount++
-		if event.Success {
-			summary.SuccessfulEvents++
-		} else {
-			summary.FailedEvents++
-		}
-		if strings.EqualFold(event.Action, "write") || strings.EqualFold(event.Action, "delete") || strings.EqualFold(event.Action, "mutate") {
-			summary.WriteEvents++
-		}
-		if event.Timestamp.Before(summary.FirstSeen) {
-			summary.FirstSeen = event.Timestamp
-		}
-		if event.Timestamp.After(summary.LastSeen) {
-			summary.LastSeen = event.Timestamp
-		}
-		summary.Targets = appendUnique(summary.Targets, event.Target)
-		summary.Environments = appendUnique(summary.Environments, event.Environment)
-		summary.Providers = appendUnique(summary.Providers, event.Provider)
-		summary.Operations[event.Operation]++
+		state.add(event)
 	}
+	report := state.report()
+	report.EventsSkipped = skipped
+	return report
+}
 
-	agents := make([]AgentSummary, 0, len(byKey))
-	for _, summary := range byKey {
-		sort.Strings(summary.Targets)
-		sort.Strings(summary.Environments)
-		sort.Strings(summary.Providers)
-		agents = append(agents, *summary)
+func addEvent(summary *AgentSummary, event Event) {
+	summary.EventCount++
+	if event.Success {
+		summary.SuccessfulEvents++
+	} else {
+		summary.FailedEvents++
+	}
+	if isWriteAction(event.Action) {
+		summary.WriteEvents++
 	}
+	if event.Timestamp.Before(summary.FirstSeen) {
+		summary.FirstSeen = event.Timestamp
+	}
+	if event.Timestamp.After(summary.LastSeen) {
+		summary.LastSeen = event.Timestamp
+	}
+	summary.Targets = appendUnique(summary.Targets, event.Target)
+	summary.Environments = appendUnique(summary.Environments, event.Environment)
+	summary.Providers = appendUnique(summary.Providers, event.Provider)
+	summary.Operations[event.Operation]++
+}
+
+func normalizeSummary(summary *AgentSummary) {
+	sort.Strings(summary.Targets)
+	sort.Strings(summary.Environments)
+	sort.Strings(summary.Providers)
+}
+
+func sortSummaries(agents []AgentSummary) {
 	sort.Slice(agents, func(i, j int) bool {
 		left := agents[i].AgentID
 		if left == "" {
@@ -62,7 +55,10 @@ func Aggregate(events []Event, skipped int) Report {
 		}
 		return left < right
 	})
-	return Report{SchemaVersion: "runtime.v1", EventsRead: len(events), EventsSkipped: skipped, Agents: agents}
+}
+
+func isWriteAction(action string) bool {
+	return strings.EqualFold(action, "write") || strings.EqualFold(action, "delete") || strings.EqualFold(action, "mutate")
 }
 
 func appendUnique(values []string, value string) []string {

internal/runtime/aggregate_stream.go

@@ -0,0 +1,72 @@
+package runtime
+
+import (
+	"bufio"
+	"encoding/json"
+	"fmt"
+	"io"
+	"strings"
+)
+
+// AggregateReader consumes JSONL one event at a time and does not retain the input events.
+func AggregateReader(r io.Reader, options Options) (Report, error) {
+	options = options.withDefaults()
+	scanner := bufio.NewScanner(r)
+	scanner.Buffer(make([]byte, 64*1024), options.MaxLineBytes)
+	state := newAggregateState()
+	line := 0
+	for scanner.Scan() {
+		line++
+		if strings.TrimSpace(scanner.Text()) == "" {
+			continue
+		}
+		if state.eventsRead >= options.MaxEvents {
+			return Report{}, fmt.Errorf("runtime event limit exceeded: %d", options.MaxEvents)
+		}
+		var event Event
+		if err := json.Unmarshal(scanner.Bytes(), &event); err != nil {
+			return Report{}, fmt.Errorf("decode runtime event %d: %w", line, err)
+		}
+		if err := validateEvent(event); err != nil {
+			return Report{}, fmt.Errorf("validate runtime event %d: %w", line, err)
+		}
+		state.add(event)
+	}
+	if err := scanner.Err(); err != nil {
+		return Report{}, fmt.Errorf("read runtime events: %w", err)
+	}
+	return state.report(), nil
+}
+
+type aggregateState struct {
+	eventsRead int
+	byKey      map[string]*AgentSummary
+}
+
+func newAggregateState() *aggregateState {
+	return &aggregateState{byKey: make(map[string]*AgentSummary)}
+}
+
+func (s *aggregateState) add(event Event) {
+	s.eventsRead++
+	key := event.AgentID
+	if key == "" {
+		key = "name:" + event.AgentName
+	}
+	summary := s.byKey[key]
+	if summary == nil {
+		summary = &AgentSummary{AgentID: event.AgentID, AgentName: event.AgentName, FirstSeen: event.Timestamp, LastSeen: event.Timestamp, Operations: make(map[string]int)}
+		s.byKey[key] = summary
+	}
+	addEvent(summary, event)
+}
+
+func (s *aggregateState) report() Report {
+	agents := make([]AgentSummary, 0, len(s.byKey))
+	for _, summary := range s.byKey {
+		normalizeSummary(summary)
+		agents = append(agents, *summary)
+	}
+	sortSummaries(agents)
+	return Report{SchemaVersion: "runtime.v1", EventsRead: s.eventsRead, Agents: agents}
+}

internal/runtime/aggregate_stream_test.go

@@ -0,0 +1,35 @@
+package runtime
+
+import (
+	"reflect"
+	"strings"
+	"testing"
+)
+
+func TestAggregateReaderMatchesAggregate(t *testing.T) {
+	input := strings.NewReader(`{"timestamp":"2026-01-02T03:04:05Z","agent_id":"a1","operation":"tool_call","target":"crm.search","provider":"crm","success":true}
+{"timestamp":"2026-01-02T03:05:05Z","agent_id":"a1","environment":"production","operation":"tool_call","target":"crm.write","provider":"crm","action":"write","success":false}
+`)
+	events, skipped, err := ReadEvents(strings.NewReader(`{"timestamp":"2026-01-02T03:04:05Z","agent_id":"a1","operation":"tool_call","target":"crm.search","provider":"crm","success":true}
+{"timestamp":"2026-01-02T03:05:05Z","agent_id":"a1","environment":"production","operation":"tool_call","target":"crm.write","provider":"crm","action":"write","success":false}
+`), Options{})
+	if err != nil {
+		t.Fatalf("ReadEvents() error = %v", err)
+	}
+	streamed, err := AggregateReader(input, Options{})
+	if err != nil {
+		t.Fatalf("AggregateReader() error = %v", err)
+	}
+	if expected := Aggregate(events, skipped); !reflect.DeepEqual(streamed, expected) {
+		t.Fatalf("streaming report differs\nstreamed=%+v\nexpected=%+v", streamed, expected)
+	}
+}
+
+func TestAggregateReaderEnforcesEventLimit(t *testing.T) {
+	input := strings.NewReader(`{"timestamp":"2026-01-02T03:04:05Z","agent_id":"a1","operation":"tool_call","target":"crm.search","success":true}
+{"timestamp":"2026-01-02T03:05:05Z","agent_id":"a1","operation":"tool_call","target":"crm.search","success":true}
+`)
+	if _, err := AggregateReader(input, Options{MaxEvents: 1}); err == nil {
+		t.Fatal("AggregateReader() did not enforce event limit")
+	}
+}

internal/runtime/audit.go

@@ -39,9 +39,11 @@ func Audit(runtimeReport Report, inventory scan.Report) AuditReport {
 			byName[name] = append(byName[name], agent)
 		}
 	}
-	providers := make(map[string]bool)
+	modelsByID := make(map[string]scan.Model, len(inventory.Models))
+	modelsByName := make(map[string]scan.Model)
 	for _, model := range inventory.Models {
-		providers[strings.ToLower(strings.TrimSpace(model.Provider))] = true
+		modelsByID[model.ID] = model
+		modelsByName[strings.ToLower(strings.TrimSpace(model.Name))] = model
 	}
 
 	for _, summary := range runtimeReport.Agents {
@@ -67,8 +69,9 @@ func Audit(runtimeReport Report, inventory scan.Report) AuditReport {
 				})
 			}
 		}
+		declaredProviders := declaredProvidersForAgent(agent, modelsByID, modelsByName)
 		for _, provider := range summary.Providers {
-			if provider != "" && !providers[strings.ToLower(strings.TrimSpace(provider))] {
+			if provider != "" && !declaredProviders[strings.ToLower(strings.TrimSpace(provider))] {
 				result.Findings = append(result.Findings, Finding{
 					RuleID: "ACP-R002", Severity: "High", Message: fmt.Sprintf("agent %q used provider %q not present in static model inventory", agent.Name, provider), AgentID: agent.ID, Confidence: 0.86,
 					Evidence: []string{provider, summary.LastSeen.Format("2006-01-02T15:04:05Z07:00")}, RemediationHint: "Verify the provider and update workspace policy or runtime configuration.",
@@ -112,6 +115,23 @@ func matchAgent(summary AgentSummary, byID map[string]scan.Agent, byName map[str
 	return scan.Agent{}, false
 }
 
+func declaredProvidersForAgent(agent scan.Agent, modelsByID map[string]scan.Model, modelsByName map[string]scan.Model) map[string]bool {
+	providers := make(map[string]bool)
+	for _, modelRef := range agent.Models {
+		model, ok := modelsByID[modelRef]
+		if !ok {
+			model, ok = modelsByName[strings.ToLower(strings.TrimSpace(modelRef))]
+		}
+		if ok {
+			provider := strings.ToLower(strings.TrimSpace(model.Provider))
+			if provider != "" {
+				providers[provider] = true
+			}
+		}
+	}
+	return providers
+}
+
 func runtimeAgentLabel(summary AgentSummary) string {
 	if summary.AgentName != "" {
 		return summary.AgentName

internal/runtime/audit_test.go

@@ -11,7 +11,10 @@ func TestAuditFindsUndeclaredProductionWrite(t *testing.T) {
 	runtimeReport := Aggregate([]Event{{
 		Timestamp: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC), AgentID: "a1", AgentName: "support", Environment: "production", Operation: "tool_call", Target: "crm.write", Provider: "unlisted", Action: "write", Success: true,
 	}}, 0)
-	result := Audit(runtimeReport, scan.Report{Agents: []scan.Agent{{ID: "a1", Name: "support", Tools: []string{"crm.search"}}}})
+	result := Audit(runtimeReport, scan.Report{
+		Agents: []scan.Agent{{ID: "a1", Name: "support", Tools: []string{"crm.search"}, Models: []string{"m1"}}},
+		Models: []scan.Model{{ID: "m1", Name: "declared-model", Provider: "declared-provider"}},
+	})
 	if result.MatchedAgents != 1 || len(result.Findings) != 3 {
 		t.Fatalf("unexpected audit result: %+v", result)
 	}