diff --git a/cmd/agentctl/main.go b/cmd/agentctl/main.go index fc46a4c..b7fa7ac 100644 --- a/cmd/agentctl/main.go +++ b/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, "", " ") diff --git a/docs/runtime-audit-mvp.md b/docs/runtime-audit-mvp.md index ac20e81..e7588ae 100644 --- a/docs/runtime-audit-mvp.md +++ b/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 --inventory `. Флаг `--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, не меняя нормализованный контракт событий. diff --git a/docs/runtime-event.v1.schema.json b/docs/runtime-event.v1.schema.json new file mode 100644 index 0000000..00b1499 --- /dev/null +++ b/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"} + } +} diff --git a/internal/runtime/adapters.go b/internal/runtime/adapters.go index 107c146..22040bd 100644 --- a/internal/runtime/adapters.go +++ b/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"` } diff --git a/internal/runtime/aggregate.go b/internal/runtime/aggregate.go index aa7d690..57d1f24 100644 --- a/internal/runtime/aggregate.go +++ b/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 { diff --git a/internal/runtime/aggregate_stream.go b/internal/runtime/aggregate_stream.go new file mode 100644 index 0000000..d446d86 --- /dev/null +++ b/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} +} diff --git a/internal/runtime/aggregate_stream_test.go b/internal/runtime/aggregate_stream_test.go new file mode 100644 index 0000000..5c9c5d3 --- /dev/null +++ b/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") + } +} diff --git a/internal/runtime/audit.go b/internal/runtime/audit.go index 9923773..9d7d1fe 100644 --- a/internal/runtime/audit.go +++ b/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 diff --git a/internal/runtime/audit_test.go b/internal/runtime/audit_test.go index 21822cb..ad52f85 100644 --- a/internal/runtime/audit_test.go +++ b/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) }