Accept autonomous run log streams

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.") {