Fix run log ingest and terminal console layout

platform/service/resources.go

@@ -328,6 +328,9 @@ func NewCoreServiceWithDurableStores(store repo.Store, logStore LogBodyStore, ar
 	for _, session := range sessions {
 		service.artifactTransfers[session.TransferID] = domain.CopyArtifactTransferSession(session)
 	}
+	if err := service.recoverJobLogStreams(); err != nil {
+		return nil, err
+	}
 	if err := service.recoverLogCursors(); err != nil {
 		return nil, err
 	}
@@ -343,6 +346,33 @@ func NewCoreServiceWithDurableStores(store repo.Store, logStore LogBodyStore, ar
 	return service, nil
 }
 
+func (svc *CoreService) recoverJobLogStreams() error {
+	jobs, err := svc.store.Jobs().List(domain.JobFilter{})
+	if err != nil {
+		return err
+	}
+	stamp := svc.now()
+	for _, job := range jobs {
+		if strings.TrimSpace(job.ID) == "" || strings.TrimSpace(job.ServerInstanceID) == "" {
+			continue
+		}
+		instance, err := svc.store.ServerInstances().Get(job.ServerInstanceID)
+		if errors.Is(err, repo.ErrNotFound) {
+			continue
+		}
+		if err != nil {
+			return err
+		}
+		if instance.State == domain.ServerInstanceStateDeleted {
+			continue
+		}
+		if err := svc.ensureJobLogStreams(job, stamp); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (svc *CoreService) recoverLogCursors() error {
 	store, ok := svc.logStore.(interface{ LatestSeq(string) (uint64, error) })
 	if !ok {
@@ -2247,6 +2277,9 @@ func (svc *CoreService) CreateJob(job domain.Job) (domain.Job, error) {
 
 	existing, err := svc.store.Jobs().GetByIdempotency(job.RunEndpointID, job.IdempotencyKey)
 	if err == nil {
+		if err := svc.ensureJobLogStreams(existing, stamp); err != nil {
+			return domain.Job{}, err
+		}
 		return existing, nil
 	}
 	if !errors.Is(err, repo.ErrNotFound) {
@@ -2283,9 +2316,57 @@ func (svc *CoreService) CreateJob(job domain.Job) (domain.Job, error) {
 	if err := svc.store.Jobs().Create(job); err != nil {
 		return domain.Job{}, err
 	}
+	if err := svc.ensureJobLogStreams(job, stamp); err != nil {
+		return domain.Job{}, err
+	}
 	return domain.CopyJob(job), nil
 }
 
+func (svc *CoreService) ensureJobLogStreams(job domain.Job, stamp time.Time) error {
+	if strings.TrimSpace(job.ServerInstanceID) == "" || strings.TrimSpace(job.ID) == "" {
+		return nil
+	}
+	streams := []struct {
+		key    string
+		source domain.LogStreamSource
+	}{
+		{key: "stdout", source: domain.LogStreamSourceProcess},
+		{key: "stderr", source: domain.LogStreamSourceProcess},
+	}
+	if job.Capability == domain.JobCapabilityRemoteRunProgram {
+		streams = append(streams,
+			struct {
+				key    string
+				source domain.LogStreamSource
+			}{key: "management-program.stdout", source: domain.LogStreamSourceManagementProgram},
+			struct {
+				key    string
+				source domain.LogStreamSource
+			}{key: "management-program.stderr", source: domain.LogStreamSourceManagementProgram},
+		)
+	}
+	for _, item := range streams {
+		stream := domain.LogStream{
+			ID:               jobLogStreamID(job.ID, item.key),
+			ServerInstanceID: job.ServerInstanceID,
+			Source:           item.source,
+			StreamKey:        item.key,
+			StorageBackend:   domain.LogStorageBackendLocalSegments,
+			RetentionPolicy:  "default",
+			CreatedAt:        stamp,
+			UpdatedAt:        stamp,
+		}
+		if _, err := svc.CreateLogStream(stream); err != nil && !errors.Is(err, repo.ErrDuplicate) {
+			return err
+		}
+	}
+	return nil
+}
+
+func jobLogStreamID(jobID string, streamKey string) string {
+	return fmt.Sprintf("job.%s.%s", jobID, streamKey)
+}
+
 func (svc *CoreService) GetJob(id string) (domain.Job, error) {
 	job, err := svc.store.Jobs().Get(id)
 	if err != nil {