diff --git a/platform/protocol/run-contracts.md b/platform/protocol/run-contracts.md index 6c5c681..f0ce1a7 100644 --- a/platform/protocol/run-contracts.md +++ b/platform/protocol/run-contracts.md @@ -84,6 +84,8 @@ Named log DTOs: Log ingest supports bounded batches, sequence ranges, checksum validation, retry-safe duplicate acknowledgement, latest sequence tracking, cursor query, and browser SSE fan-out from already-ingested platform logs. Log payloads must not carry artifact chunks, host paths, raw credentials, direct sockets, or unbounded inline data. +Run-assigned Platform jobs use `job..` log stream IDs. Autonomous lifecycle bootstrap is Run-owned machine execution rather than a Platform job, so its durable process logs use `run...`. Platform may auto-create those streams only after validating the active Run session and the server-to-Run binding. For retry compatibility, legacy spooled `job.autonomous-*.` batches are accepted as Run-owned autonomous streams without creating or completing a Platform job. + Log ingest is durable and independently retried. Artifact/file transfer backlog must not prevent log batch acknowledgement, duplicate acknowledgement, cursor state updates, or spool cleanup. The platform stores log stream metadata through `repo.Store` and stores log bodies through the configured `LogBodyStore`. The default `file` backend persists platform metadata to `PLATFORM_METADATA_PATH` and appends log entries to segmented JSONL files under `PLATFORM_LOG_DIR`; the `memory` backend is only for tests and disposable local development. MySQL/Postgres are appropriate for platform metadata, stream state, retention policy, indexes, and audit records, but should not be the primary row-per-log-line store for hundreds or thousands of servers. Production log bodies should move behind the same boundary to append/query backends such as ClickHouse, Loki, OpenSearch/Elasticsearch, or object-storage segments with compact indexes. diff --git a/platform/service/log_ingest.go b/platform/service/log_ingest.go index c62218f..95b7095 100644 --- a/platform/service/log_ingest.go +++ b/platform/service/log_ingest.go @@ -25,7 +25,7 @@ func (svc *CoreService) IngestLogBatch(batch domain.LogBatchIngest) (domain.LogB stamp := svc.now() stream, err := svc.store.LogStreams().Get(batch.LogStreamID) if errors.Is(err, repo.ErrNotFound) { - if repairErr := svc.ensureJobLogStreamForBatch(batch, stamp); repairErr == nil { + if repairErr := svc.ensureLogStreamForBatch(batch, stamp); repairErr == nil { stream, err = svc.store.LogStreams().Get(batch.LogStreamID) } } @@ -112,6 +112,49 @@ func (svc *CoreService) ensureJobLogStreamForBatch(batch domain.LogBatchIngest, return svc.ensureJobLogStreams(job, stamp) } +func (svc *CoreService) ensureLogStreamForBatch(batch domain.LogBatchIngest, stamp time.Time) error { + jobID, hasJobID := jobIDFromLogBatch(batch) + if hasJobID { + job, err := svc.store.Jobs().Get(jobID) + if err == nil { + if job.ServerInstanceID != batch.ServerInstanceID || job.RunEndpointID != batch.RunEndpointID { + return validationError("log batch job scope does not match stream") + } + return svc.ensureJobLogStreams(job, stamp) + } + if !errors.Is(err, repo.ErrNotFound) || !strings.HasPrefix(jobID, "autonomous-") { + return err + } + } + return svc.ensureRunLogStreamForBatch(batch, stamp) +} + +func (svc *CoreService) ensureRunLogStreamForBatch(batch domain.LogBatchIngest, stamp time.Time) error { + if batch.Source != domain.LogStreamSourceProcess && batch.Source != domain.LogStreamSourceManagementProgram { + return repo.ErrNotFound + } + if batch.LogStreamID != runLogStreamID(batch.RunEndpointID, batch.ServerInstanceID, batch.StreamKey) && !legacyAutonomousLogStream(batch) { + return repo.ErrNotFound + } + instance, err := svc.store.ServerInstances().Get(batch.ServerInstanceID) + if err != nil { + return err + } + if instance.RunEndpointID != batch.RunEndpointID { + return validationError("server instance run endpoint must match log batch endpoint") + } + _, err = svc.CreateLogStream(domain.LogStream{ID: batch.LogStreamID, ServerInstanceID: batch.ServerInstanceID, Source: batch.Source, StreamKey: batch.StreamKey, StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default", CreatedAt: stamp, UpdatedAt: stamp}) + if errors.Is(err, repo.ErrDuplicate) { + return nil + } + return err +} + +func legacyAutonomousLogStream(batch domain.LogBatchIngest) bool { + jobID, ok := jobIDFromLogBatch(batch) + return ok && strings.HasPrefix(jobID, "autonomous-") +} + func jobIDFromLogBatch(batch domain.LogBatchIngest) (string, bool) { streamKey := strings.TrimSpace(batch.StreamKey) if streamKey == "" || !strings.HasPrefix(batch.LogStreamID, "job.") { diff --git a/platform/service/log_ingest_test.go b/platform/service/log_ingest_test.go index d93f2f0..d2d3d3b 100644 --- a/platform/service/log_ingest_test.go +++ b/platform/service/log_ingest_test.go @@ -233,6 +233,63 @@ func TestCoreServiceRepairsMissingDeclaredProcessLogStreamOnIngest(t *testing.T) } } +func TestCoreServiceAcceptsAutonomousRunLogStreamWithoutPlatformJob(t *testing.T) { + svc, sessionToken := newRegisteredLogIngestService(t) + entry := domain.LogEntry{Seq: 1, Timestamp: time.Date(2026, 7, 3, 12, 0, 1, 0, time.UTC), Level: "info", Line: "autonomous bootstrap output"} + streamID := runLogStreamID("run-local", "server-1", "scum.console.stdout") + ack, err := svc.IngestLogBatch(domain.LogBatchIngest{ + RunEndpointID: "run-local", + SessionToken: sessionToken, + LogStreamID: streamID, + ServerInstanceID: "server-1", + StreamKey: "scum.console.stdout", + Source: domain.LogStreamSourceProcess, + FirstSeq: entry.Seq, + LastSeq: entry.Seq, + Compression: "none", + Checksum: validator.LogLineChecksum(entry.Line), + Entries: []domain.LogEntry{entry}, + }) + if err != nil { + t.Fatalf("ingest autonomous run log batch: %v", err) + } + if !ack.Accepted || ack.LogStreamID != streamID || ack.LatestSeq != entry.Seq { + t.Fatalf("unexpected autonomous stream ack: %+v", ack) + } + stream, err := svc.GetLogStream(streamID) + if err != nil { + t.Fatalf("get autonomous stream: %v", err) + } + if stream.ServerInstanceID != "server-1" || stream.StreamKey != "scum.console.stdout" || stream.Source != domain.LogStreamSourceProcess { + t.Fatalf("unexpected autonomous stream metadata: %+v", stream) + } +} + +func TestCoreServiceAcceptsLegacyAutonomousJobLogStreamWithoutPlatformJob(t *testing.T) { + svc, sessionToken := newRegisteredLogIngestService(t) + entry := domain.LogEntry{Seq: 1, Timestamp: time.Date(2026, 7, 3, 12, 0, 1, 0, time.UTC), Level: "info", Line: "legacy autonomous bootstrap output"} + streamID := jobLogStreamID("autonomous-bootstrap-start", "scum.console.stdout") + ack, err := svc.IngestLogBatch(domain.LogBatchIngest{ + RunEndpointID: "run-local", + SessionToken: sessionToken, + LogStreamID: streamID, + ServerInstanceID: "server-1", + StreamKey: "scum.console.stdout", + Source: domain.LogStreamSourceProcess, + FirstSeq: entry.Seq, + LastSeq: entry.Seq, + Compression: "none", + Checksum: validator.LogLineChecksum(entry.Line), + Entries: []domain.LogEntry{entry}, + }) + if err != nil { + t.Fatalf("ingest legacy autonomous log batch: %v", err) + } + if !ack.Accepted || ack.LogStreamID != streamID || ack.LatestSeq != entry.Seq { + t.Fatalf("unexpected legacy autonomous stream ack: %+v", ack) + } +} + func TestCoreServiceRejectsOutOfOrderAndConflictingLogBatches(t *testing.T) { svc, sessionToken := newRegisteredLogIngestService(t) createLogStreamFixture(t, svc) diff --git a/platform/service/resources.go b/platform/service/resources.go index 99a3a2a..505aa41 100644 --- a/platform/service/resources.go +++ b/platform/service/resources.go @@ -2543,6 +2543,10 @@ func jobLogStreamID(jobID string, streamKey string) string { return fmt.Sprintf("job.%s.%s", jobID, streamKey) } +func runLogStreamID(runEndpointID string, serverInstanceID string, streamKey string) string { + return fmt.Sprintf("run.%s.%s.%s", runEndpointID, serverInstanceID, streamKey) +} + func (svc *CoreService) GetJob(id string) (domain.Job, error) { job, err := svc.store.Jobs().Get(id) if err != nil {