Keep SCUM poll jobs and drop orphan job log streams

The metadata snapshot deleted every SCUM sqlite query job while persisting, so
the recurring database poll lost its job records as soon as anything wrote the
snapshot and left its stdout/stderr streams behind. Those polls are the only
producer for the platform scum_user and scum_vehicle tables, and the service
already retires older terminal polls with their streams, so the snapshot no
longer drops them.

Startup now removes job log streams whose job no longer exists, keeping the
autonomous lifecycle streams the Run posts without a platform job. That clears
the streams left by the removed poll producers instead of carrying them in
every snapshot.

platform/service/resources.go

@@ -336,6 +336,9 @@ func NewCoreServiceWithDurableStores(store repo.Store, logStore LogBodyStore, ar
 	if err := service.recoverJobLogStreams(); err != nil {
 		return nil, err
 	}
+	if err := service.pruneOrphanJobLogStreams(); err != nil {
+		return nil, err
+	}
 	if err := service.recoverLogCursors(); err != nil {
 		return nil, err
 	}
@@ -375,6 +378,52 @@ func (svc *CoreService) recoverJobLogStreams() error {
 	return nil
 }
 
+// pruneOrphanJobLogStreams drops job-scoped log streams whose job no longer
+// exists. Deleting a job removes its own streams, so this only clears leftovers
+// from producers that are gone. It runs once at startup because the stream
+// table is otherwise written one job at a time.
+func (svc *CoreService) pruneOrphanJobLogStreams() error {
+	jobs, err := svc.store.Jobs().List(domain.JobFilter{})
+	if err != nil {
+		return err
+	}
+	known := make(map[string]struct{}, len(jobs))
+	for _, job := range jobs {
+		known[job.ID] = struct{}{}
+	}
+	streams, err := svc.store.LogStreams().List(domain.LogStreamFilter{})
+	if err != nil {
+		return err
+	}
+	for _, stream := range streams {
+		jobID, ok := jobIDFromJobLogStream(stream)
+		if !ok || strings.HasPrefix(jobID, "autonomous-") {
+			continue
+		}
+		if _, exists := known[jobID]; exists {
+			continue
+		}
+		if err := svc.store.LogStreams().Delete(stream.ID); err != nil && !errors.Is(err, repo.ErrNotFound) {
+			return err
+		}
+	}
+	return nil
+}
+
+func jobIDFromJobLogStream(stream domain.LogStream) (string, bool) {
+	streamKey := strings.TrimSpace(stream.StreamKey)
+	if streamKey == "" || !strings.HasPrefix(stream.ID, "job.") {
+		return "", false
+	}
+	body := strings.TrimPrefix(stream.ID, "job.")
+	suffix := "." + streamKey
+	if !strings.HasSuffix(body, suffix) {
+		return "", false
+	}
+	jobID := strings.TrimSuffix(body, suffix)
+	return jobID, strings.TrimSpace(jobID) != ""
+}
+
 func (svc *CoreService) recoverLogCursors() error {
 	store, ok := svc.logStore.(interface{ LatestSeq(string) (uint64, error) })
 	if !ok {

platform/service/resources_test.go

@@ -282,6 +282,79 @@ func TestCoreServiceStartupRecoversLegacyJobLogStreams(t *testing.T) {
 	}
 }
 
+func TestCoreServiceStartupPrunesOrphanJobLogStreams(t *testing.T) {
+	store := repo.NewMemoryStore()
+	seed := newCoreService(store, func() time.Time { return fixedTime })
+	plugin, endpoint := createPluginAndRunEndpoint(t, seed)
+	plugin.RequiredRunCapabilities = append(plugin.RequiredRunCapabilities, domain.JobCapabilityRemoteRunProgram)
+	if err := seed.store.GamePlugins().Update(plugin); err != nil {
+		t.Fatalf("update plugin capabilities: %v", err)
+	}
+	endpoint.Capabilities = append(endpoint.Capabilities, domain.JobCapabilityRemoteRunProgram)
+	if err := seed.store.RunEndpoints().Update(endpoint); err != nil {
+		t.Fatalf("update endpoint capabilities: %v", err)
+	}
+	if _, err := seed.CreateServerInstance(domain.ServerInstance{ID: "orphan-stream-server", PluginID: plugin.ID, RunEndpointID: endpoint.ID, Name: "Orphan Streams"}); err != nil {
+		t.Fatalf("create server instance: %v", err)
+	}
+	if err := store.Jobs().Create(domain.Job{
+		ID:               "live-terminal-job",
+		ServerInstanceID: "orphan-stream-server",
+		RunEndpointID:    endpoint.ID,
+		Capability:       domain.JobCapabilityRemoteRunProgram,
+		TargetKey:        "protected-program",
+		InputRef:         "input://protected-program/live-terminal-job",
+		IdempotencyKey:   "live-terminal",
+		State:            domain.JobStateQueued,
+		CreatedAt:        fixedTime,
+		UpdatedAt:        fixedTime,
+	}); err != nil {
+		t.Fatalf("seed live job: %v", err)
+	}
+	orphanStream := domain.LogStream{
+		ID:               jobLogStreamID("job-plugin-query-poll-gone", "stdout"),
+		ServerInstanceID: "orphan-stream-server",
+		Source:           domain.LogStreamSourceProcess,
+		StreamKey:        "stdout",
+		StorageBackend:   domain.LogStorageBackendLocalSegments,
+		RetentionPolicy:  "default",
+	}
+	autonomousStream := domain.LogStream{
+		ID:               jobLogStreamID("autonomous-bootstrap-start", "scum.console.stdout"),
+		ServerInstanceID: "orphan-stream-server",
+		Source:           domain.LogStreamSourceProcess,
+		StreamKey:        "scum.console.stdout",
+		StorageBackend:   domain.LogStorageBackendLocalSegments,
+		RetentionPolicy:  "default",
+	}
+	for _, stream := range []domain.LogStream{orphanStream, autonomousStream} {
+		if err := store.LogStreams().Create(stream); err != nil {
+			t.Fatalf("seed log stream %s: %v", stream.ID, err)
+		}
+	}
+	recovered, err := NewCoreServiceWithDurableStores(store, NewMemoryLogBodyStore(), NewMemoryArtifactBodyStore())
+	if err != nil {
+		t.Fatalf("recover durable service: %v", err)
+	}
+	streams, err := recovered.ListLogStreams(domain.LogStreamFilter{ServerInstanceID: "orphan-stream-server"})
+	if err != nil {
+		t.Fatalf("list recovered streams: %v", err)
+	}
+	byID := map[string]domain.LogStream{}
+	for _, stream := range streams {
+		byID[stream.ID] = stream
+	}
+	if _, exists := byID[orphanStream.ID]; exists {
+		t.Fatalf("expected orphan job log stream to be pruned, streams=%+v", byID)
+	}
+	if _, exists := byID[autonomousStream.ID]; !exists {
+		t.Fatalf("expected autonomous job log stream to survive, streams=%+v", byID)
+	}
+	if _, exists := byID[jobLogStreamID("live-terminal-job", "stdout")]; !exists {
+		t.Fatalf("expected live job log stream to be recovered, streams=%+v", byID)
+	}
+}
+
 func TestCoreServiceRejectsInvalidServerDependencies(t *testing.T) {
 	svc := newTestCoreService()
 	plugin, endpoint := createPluginAndRunEndpoint(t, svc)