diff --git a/platform/service/log_ingest_test.go b/platform/service/log_ingest_test.go index 27824de..f2a0e30 100644 --- a/platform/service/log_ingest_test.go +++ b/platform/service/log_ingest_test.go @@ -126,6 +126,30 @@ func TestCoreServicePersistsAndEnforcesImmutableProcessLogSessionMetadata(t *tes } } +func TestCoreServiceAcceptsLegacyProcessStreamStartWithoutSessionID(t *testing.T) { + svc, sessionToken := newRegisteredLogIngestService(t) + startedAt := time.Date(2026, 7, 3, 12, 30, 0, 0, time.UTC) + streamID := "run.run-local.server-1.stdout" + if _, err := svc.CreateLogStream(domain.LogStream{ + ID: streamID, + ServerInstanceID: "server-1", + Source: domain.LogStreamSourceProcess, + StreamKey: "stdout", + SessionStartedAt: startedAt, + StorageBackend: domain.LogStorageBackendLocalSegments, + RetentionPolicy: "default", + }); err != nil { + t.Fatalf("create legacy process stream: %v", err) + } + + batch := validLogBatch(t, sessionToken, 1, 1) + batch.LogStreamID = streamID + batch.SessionStartedAt = startedAt + if _, err := svc.IngestLogBatch(batch); err != nil { + t.Fatalf("ingest legacy process batch: %v", err) + } +} + func TestCoreServiceSerializesConsistentSessionStreamCreation(t *testing.T) { svc, _ := newRegisteredLogIngestService(t) startedAt := time.Date(2026, 7, 3, 12, 30, 0, 0, time.UTC) diff --git a/platform/validator/log_ingest.go b/platform/validator/log_ingest.go index d4b478b..2da4837 100644 --- a/platform/validator/log_ingest.go +++ b/platform/validator/log_ingest.go @@ -29,7 +29,10 @@ func ValidateLogBatchIngest(batch domain.LogBatchIngest) error { } hasSessionID := strings.TrimSpace(batch.LogSessionID) != "" hasSessionStart := !batch.SessionStartedAt.IsZero() - if hasSessionID != hasSessionStart { + // Older process streams persisted their start time before Run assigned a + // generation id. Keep those batches ingestible so a durable Run spool can + // drain without blocking newer generation-scoped console output. + if hasSessionID && !hasSessionStart { violations = append(violations, "logSessionId and sessionStartedAt must be provided together") } if (hasSessionID || hasSessionStart) && batch.Source != domain.LogStreamSourceProcess { diff --git a/platform/validator/resources.go b/platform/validator/resources.go index 1ec7dac..85b4a22 100644 --- a/platform/validator/resources.go +++ b/platform/validator/resources.go @@ -1651,7 +1651,9 @@ func ValidateLogStream(stream domain.LogStream) error { } hasSessionID := strings.TrimSpace(stream.LogSessionID) != "" hasSessionStart := !stream.SessionStartedAt.IsZero() - if hasSessionID != hasSessionStart { + // Keep legacy process streams with only a persisted start time valid while + // requiring the complete pair for new generation-scoped streams. + if hasSessionID && !hasSessionStart { violations = append(violations, "logSessionId and sessionStartedAt must be provided together") } if (hasSessionID || hasSessionStart) && stream.Source != domain.LogStreamSourceProcess {