Freeze SCUM log-source tailing run contract

platform/service/distributions.go

@@ -764,7 +764,7 @@ func (svc *CoreService) QueueLogBackfillForSession(sessionID string, request dom
 		InputRef:         request.CheckpointRef,
 		IdempotencyKey:   request.IdempotencyKey,
 		Progress:         domain.JobProgress{Percent: 0, Message: "historical log backfill queued"},
-		ExecutionInput:   domain.JobExecutionInput{LogSource: &source},
+		ExecutionInput:   domain.JobExecutionInput{PluginID: plugin.ID, TargetVersion: plugin.Version, LogSource: &source},
 	})
 	if err != nil {
 		_ = svc.recordAuditEvent(user.ID, "logs.backfill.denied", "server-instance", instance.ID, domain.AuditResultDenied, "log backfill denied: endpoint unsupported or offline")

platform/service/job_channel.go

@@ -288,6 +288,9 @@ func (svc *CoreService) CompleteRunJob(result domain.RunJobResult) (domain.RunJo
 }
 
 func validateExecutionResultForJob(job domain.Job, result domain.RunJobResult) error {
+	if result.ExecutionResult.ParsedLogBatch != nil && job.Capability != domain.JobCapabilityLogsBackfill {
+		return validationError("parsed log batch result is allowed only for logs.backfill jobs")
+	}
 	if definition := job.ExecutionInput.Deployment; definition != nil {
 		receipt := result.ExecutionResult.DeploymentReceipt
 		if result.State == domain.JobStateSucceeded && definition.Mode == domain.ServerDeploymentModeCustom && receipt == nil {
@@ -388,6 +391,18 @@ func validateExecutionResultForJob(job domain.Job, result domain.RunJobResult) e
 				return err
 			}
 		}
+	case domain.JobCapabilityLogsBackfill:
+		if result.ExecutionResult.ParsedLogBatch == nil {
+			break
+		}
+		if result.ExecutionResult.Kind != "" && result.ExecutionResult.Kind != scumParsedLogBatchExecutionKind {
+			return validationError("parsed log batch result type is invalid")
+		}
+		if result.State == domain.JobStateSucceeded {
+			if err := validateSCUMParsedLogBatchResultForJob(job, *result.ExecutionResult.ParsedLogBatch); err != nil {
+				return err
+			}
+		}
 	case domain.JobCapabilityClientManagerDeploy:
 		if result.State == domain.JobStateSucceeded && result.ExecutionResult.Kind != "client-manager.deployed" {
 			return validationError("client-manager deploy result type is invalid")
@@ -787,7 +802,11 @@ func terminalFingerprint(result domain.RunJobResult) string {
 	if result.ExecutionResult.GuardedMutation != nil {
 		guardedMutationFingerprint = fmt.Sprintf("%s|%s|%s|%d|%s", result.ExecutionResult.GuardedMutation.ResultDigest, result.ExecutionResult.GuardedMutation.AssetDigest, result.ExecutionResult.GuardedMutation.PatchDigest, result.ExecutionResult.GuardedMutation.AffectedRows, result.ExecutionResult.GuardedMutation.ReadbackStatus)
 	}
-	return fmt.Sprintf("%s|%d|%s|%s|%s|%s|%t|%s|%s|%s|%s|%s|%s", result.State, result.Progress.Percent, result.ResultRef, result.Message, result.ErrorCode, result.Progress.Message, result.Retryable, result.ExecutionResult.Kind, result.ExecutionResult.Checksum, schemaProbeFingerprint, sqliteTemplateFingerprint, rconTemplateFingerprint, guardedMutationFingerprint)
+	parsedLogBatchFingerprint := ""
+	if result.ExecutionResult.ParsedLogBatch != nil {
+		parsedLogBatchFingerprint = fmt.Sprintf("%s|%s|%s|%s|%d|%s", result.ExecutionResult.ParsedLogBatch.ResultDigest, result.ExecutionResult.ParsedLogBatch.AssetDigest, result.ExecutionResult.ParsedLogBatch.ParserDigest, result.ExecutionResult.ParsedLogBatch.FirstCursor.StreamGeneration, result.ExecutionResult.ParsedLogBatch.EventCount, result.ExecutionResult.ParsedLogBatch.TailState)
+	}
+	return fmt.Sprintf("%s|%d|%s|%s|%s|%s|%t|%s|%s|%s|%s|%s|%s|%s", result.State, result.Progress.Percent, result.ResultRef, result.Message, result.ErrorCode, result.Progress.Message, result.Retryable, result.ExecutionResult.Kind, result.ExecutionResult.Checksum, schemaProbeFingerprint, sqliteTemplateFingerprint, rconTemplateFingerprint, guardedMutationFingerprint, parsedLogBatchFingerprint)
 }
 
 func terminalMessage(result domain.RunJobResult) string {

platform/service/job_channel_test.go

@@ -320,6 +320,40 @@ func TestCoreServiceRunJobGuardedMutationEnvelopeIsFencedToLease(t *testing.T) {
 	}
 }
 
+func TestCoreServiceRunJobParsedLogBatchEnvelopeIsFencedToLease(t *testing.T) {
+	svc, sessionToken := newRegisteredRunJobService(t)
+	binding := scumTemplateTestBinding()
+	createSCUMTemplateServerFixture(t, svc, binding)
+	source := &domain.RuntimeLogSource{Key: "scum-login-events", Kind: "file.tail", TargetKey: "logs/login", StreamKey: "scum.login", CursorKind: "fingerprint", RetentionDays: 90}
+	job, err := svc.CreateJob(domain.Job{ID: "job-log", ServerInstanceID: binding.ServerInstanceID, RunEndpointID: binding.RunEndpointID, Capability: domain.JobCapabilityLogsBackfill, TargetKey: "logs/scum-login-events", InputRef: "artifact://logs/checkpoint/1", IdempotencyKey: "idem-log", ExecutionInput: domain.JobExecutionInput{PluginID: binding.PluginID, TargetVersion: binding.PluginVersion, LogSource: source, Inputs: map[string]string{"parserKey": "scum-login-log-login-parser", "parserVersion": "scum-login-log-v1", "parserDigest": scumTemplateTestHash(), "adapterVersion": binding.AdapterVersion}}})
+	if err != nil {
+		t.Fatalf("create parsed log backfill job: %v", err)
+	}
+	claim, err := svc.ClaimRunJob(domain.RunJobClaim{RunEndpointID: binding.RunEndpointID, SessionToken: sessionToken, Capabilities: []string{domain.JobCapabilityLogsBackfill}, Capacity: domain.RunCapacity{MaxJobs: 4}})
+	if err != nil || !claim.HasJob || claim.Job.JobID != job.ID || claim.Job.ExecutionInput.LogSource == nil {
+		t.Fatalf("claim parsed log job: claim=%+v err=%v", claim, err)
+	}
+	result := scumParsedLogBatchTestResult(binding, job.ID, *source)
+	if _, err := svc.CompleteRunJob(domain.RunJobResult{RunEndpointID: binding.RunEndpointID, SessionToken: sessionToken, JobID: claim.Job.JobID, LeaseToken: claim.Job.LeaseToken, Attempt: claim.Job.Attempt, State: domain.JobStateSucceeded, Progress: domain.RunJobProgressReport{Percent: 100, Message: "log parse complete"}, Message: "log parse complete", ExecutionResult: domain.JobExecutionResult{Kind: scumParsedLogBatchExecutionKind, ParsedLogBatch: &result, AuditSummary: "redacted parsed log batch"}}); err != nil {
+		t.Fatalf("complete matching parsed log job: %v", err)
+	}
+
+	badJob, err := svc.CreateJob(domain.Job{ID: "job-log-bad", ServerInstanceID: binding.ServerInstanceID, RunEndpointID: binding.RunEndpointID, Capability: domain.JobCapabilityLogsBackfill, TargetKey: "logs/scum-login-events", InputRef: "artifact://logs/checkpoint/2", IdempotencyKey: "idem-log-bad", ExecutionInput: domain.JobExecutionInput{PluginID: binding.PluginID, TargetVersion: binding.PluginVersion, LogSource: source, Inputs: map[string]string{"parserKey": "scum-login-log-login-parser", "parserVersion": "scum-login-log-v1", "parserDigest": scumTemplateTestHash(), "adapterVersion": binding.AdapterVersion}}})
+	if err != nil {
+		t.Fatalf("create bad parsed log job: %v", err)
+	}
+	badClaim, err := svc.ClaimRunJob(domain.RunJobClaim{RunEndpointID: binding.RunEndpointID, SessionToken: sessionToken, Capabilities: []string{domain.JobCapabilityLogsBackfill}, Capacity: domain.RunCapacity{MaxJobs: 4}})
+	if err != nil || !badClaim.HasJob || badClaim.Job.JobID != badJob.ID {
+		t.Fatalf("claim bad parsed log job: claim=%+v err=%v", badClaim, err)
+	}
+	badResult := scumParsedLogBatchTestResult(binding, badJob.ID, *source)
+	badResult.ParserDigest = "sha256:" + strings.Repeat("b", 64)
+	_, err = svc.CompleteRunJob(domain.RunJobResult{RunEndpointID: binding.RunEndpointID, SessionToken: sessionToken, JobID: badClaim.Job.JobID, LeaseToken: badClaim.Job.LeaseToken, Attempt: badClaim.Job.Attempt, State: domain.JobStateSucceeded, Progress: domain.RunJobProgressReport{Percent: 100, Message: "log parse complete"}, Message: "log parse complete", ExecutionResult: domain.JobExecutionResult{Kind: scumParsedLogBatchExecutionKind, ParsedLogBatch: &badResult, AuditSummary: "redacted parsed log batch"}})
+	if err == nil || !strings.Contains(err.Error(), "parser identity") {
+		t.Fatalf("expected parser-fenced parsed log result rejection, got %v", err)
+	}
+}
+
 func TestCoreServiceRunJobReconcile(t *testing.T) {
 	svc, sessionToken := newRegisteredRunJobService(t)
 	createQueuedRunJob(t, svc, "job-1", "idem-1")
@@ -350,7 +384,7 @@ func newRegisteredRunJobService(t *testing.T) (*CoreService, string) {
 	t.Helper()
 	svc := newTestCoreService()
 	helloRequest := validRunControlHello()
-	helloRequest.CapabilityReport.Capabilities = append(helloRequest.CapabilityReport.Capabilities, "process.start", domain.JobCapabilityRemoteRunDBSQLiteQuery, domain.JobCapabilityRemoteRunProtectedRCON, domain.JobCapabilityRemoteRunProtectedSQL)
+	helloRequest.CapabilityReport.Capabilities = append(helloRequest.CapabilityReport.Capabilities, "process.start", domain.JobCapabilityRemoteRunDBSQLiteQuery, domain.JobCapabilityRemoteRunProtectedRCON, domain.JobCapabilityRemoteRunProtectedSQL, domain.JobCapabilityLogsBackfill)
 	helloRequest.CapabilityReport.Fingerprint = "cap-jobs"
 	hello, err := svc.RegisterRunHello(helloRequest)
 	if err != nil {
@@ -379,7 +413,7 @@ func scumTypedRCONSourcePlan() *domain.RuntimeSourceRCONPlan {
 
 func createSCUMTemplateServerFixture(t *testing.T, svc *CoreService, binding domain.SCUMBindingIdentity) {
 	t.Helper()
-	if _, err := svc.CreateGamePlugin(domain.GamePlugin{ID: binding.PluginID, Name: "SCUM", Version: binding.PluginVersion, ServerType: "scum", ManifestRef: "artifact://manifests/game.scum/0.1.6", CreateFormSchemaRef: "artifact://schemas/game.scum/create-form/0.1.6", RequiredRunCapabilities: []string{domain.JobCapabilityRemoteRunDBSQLiteQuery, domain.JobCapabilityRemoteRunProtectedRCON, domain.JobCapabilityRemoteRunProtectedSQL}, DeclaredPermissions: []string{"server.remote.access", "server.game-client.command", "server.game-client.maintenance"}, Permissions: domain.PluginPermissions{RemoteAccess: true}}); err != nil {
+	if _, err := svc.CreateGamePlugin(domain.GamePlugin{ID: binding.PluginID, Name: "SCUM", Version: binding.PluginVersion, ServerType: "scum", ManifestRef: "artifact://manifests/game.scum/0.1.6", CreateFormSchemaRef: "artifact://schemas/game.scum/create-form/0.1.6", RequiredRunCapabilities: []string{domain.JobCapabilityRemoteRunDBSQLiteQuery, domain.JobCapabilityRemoteRunProtectedRCON, domain.JobCapabilityRemoteRunProtectedSQL, domain.JobCapabilityLogsBackfill}, DeclaredPermissions: []string{"server.remote.access", "server.game-client.command", "server.game-client.maintenance", "server.logs.read"}, Permissions: domain.PluginPermissions{RemoteAccess: true}}); err != nil {
 		t.Fatalf("create SCUM plugin fixture: %v", err)
 	}
 	if _, err := svc.CreateServerInstance(domain.ServerInstance{ID: binding.ServerInstanceID, PluginID: binding.PluginID, RunEndpointID: binding.RunEndpointID, Name: "SCUM"}); err != nil {
@@ -399,6 +433,11 @@ func scumGuardedMutationTestResult(request domain.SCUMGuardedMutationRequest) do
 	return domain.SCUMGuardedMutationResult{RequestID: request.RequestID, JobID: request.JobID, Binding: request.Binding, Status: domain.SCUMTerminalResultSucceeded, Capability: request.Capability, TargetKey: request.TargetKey, TemplateKey: request.TemplateKey, AdapterVersion: request.AdapterVersion, SchemaFingerprint: request.RequiredSchemaFingerprint, AssetDigest: request.AssetDigest, SourceFingerprint: scumTemplateTestHash(), TargetIdentityDigest: request.TargetIdentityDigest, ExpectedRowDigest: request.ExpectedRowDigest, ExpectedValueDigest: request.ExpectedValueDigest, ExpectedXMLDigest: request.ExpectedXMLDigest, PatchDigest: request.PatchDigest, BackupEvidenceDigest: request.BackupEvidenceDigest, OfflineEvidenceDigest: request.OfflineEvidenceDigest, DangerConfirmationDigest: request.DangerConfirmationDigest, ReadbackExpectationDigest: request.ReadbackExpectationDigest, ObservedAt: time.Now(), ResultDigest: scumTemplateTestHash(), BeforeDigest: scumTemplateTestHash(), AfterDigest: scumTemplateTestHash(), ReadbackDigest: scumTemplateTestHash(), AffectedRows: 1, ReadbackStatus: domain.SCUMMutationReadbackConfirmed, SafeSummary: "confirmed by declared readback", Limits: request.Bounds, SafeError: domain.SCUMSafeError{Code: domain.SCUMSafeErrorNone}}
 }
 
+func scumParsedLogBatchTestResult(binding domain.SCUMBindingIdentity, jobID string, source domain.RuntimeLogSource) domain.SCUMParsedLogBatchResult {
+	cursor := domain.SCUMParsedLogCursor{SourceIdentityDigest: scumTemplateTestHash(), StreamGeneration: scumTemplateTestHash(), Sequence: 7}
+	return domain.SCUMParsedLogBatchResult{RequestID: "request-log", JobID: jobID, Binding: binding, Status: domain.SCUMTerminalResultSucceeded, SourceKey: source.Key, StreamKey: source.StreamKey, ParserKey: "scum-login-log-login-parser", ParserVersion: "scum-login-log-v1", AdapterVersion: binding.AdapterVersion, AssetDigest: scumTemplateTestHash(), ParserDigest: scumTemplateTestHash(), ObservedAt: time.Now(), ResultDigest: scumTemplateTestHash(), FirstCursor: cursor, LastCursor: cursor, TailState: domain.SCUMLogTailRotated, Replay: true, EventCount: 1, Events: []domain.SCUMParsedLogEvent{{EventType: "scum.login", OccurredAt: time.Now(), Cursor: cursor, LogicalEventDigest: scumTemplateTestHash(), EventDigest: scumTemplateTestHash(), PayloadDigest: scumTemplateTestHash(), Payload: map[string]any{"externalPlayerId": "player-redacted", "displayName": "Known Player", "profileLocalId": "profile-redacted"}}}, SafeSummary: "one sanitized login event parsed from declared source", Limits: domain.DefaultSCUMParsedLogBatchBounds(), SafeError: domain.SCUMSafeError{Code: domain.SCUMSafeErrorNone}}
+}
+
 func createQueuedRunJob(t *testing.T, svc *CoreService, id string, idempotencyKey string) domain.Job {
 	t.Helper()
 	job, err := svc.CreateJob(domain.Job{

platform/service/scum_live_data.go

@@ -14,6 +14,7 @@ const (
 	scumSQLiteTemplateExecutionKind  = "sqlite.template-query"
 	scumRCONTemplateExecutionKind    = "rcon.template-command"
 	scumGuardedMutationExecutionKind = "sqlite.guarded-mutation"
+	scumParsedLogBatchExecutionKind  = "log.parsed-events"
 )
 
 func (svc *CoreService) RequestSCUMSchemaProbeForSession(sessionID, serverInstanceID, idempotencyKey string) (domain.SCUMSchemaProbeRequest, domain.RemoteAdapterResult, error) {
@@ -217,6 +218,45 @@ func validateSCUMGuardedMutationResultForJob(job domain.Job, result domain.SCUMG
 	return nil
 }
 
+func validateSCUMParsedLogBatchResultForJob(job domain.Job, result domain.SCUMParsedLogBatchResult) error {
+	if err := validator.ValidateSCUMParsedLogBatchResult(result); err != nil {
+		return err
+	}
+	expected := job.ExecutionInput.LogSource
+	if expected == nil {
+		return validationError("parsed log batch request is missing from leased job")
+	}
+	if result.JobID != job.ID {
+		return validationError("parsed log batch result does not match leased job identity")
+	}
+	if result.Binding.ServerInstanceID != job.ServerInstanceID || result.Binding.RunEndpointID != job.RunEndpointID {
+		return validationError("parsed log batch result does not match leased server or Run endpoint")
+	}
+	if result.SourceKey != expected.Key || result.StreamKey != expected.StreamKey {
+		return validationError("parsed log batch result does not match leased log source identity")
+	}
+	for key, value := range map[string]string{
+		"parserKey":      result.ParserKey,
+		"parserVersion":  result.ParserVersion,
+		"parserDigest":   result.ParserDigest,
+		"adapterVersion": result.AdapterVersion,
+	} {
+		if expectedValue := strings.TrimSpace(job.ExecutionInput.Inputs[key]); expectedValue != "" && value != expectedValue {
+			return validationError("parsed log batch result does not match leased parser identity")
+		}
+	}
+	if job.ExecutionInput.PluginID != "" && result.Binding.PluginID != job.ExecutionInput.PluginID {
+		return validationError("parsed log batch result does not match leased plugin identity")
+	}
+	if job.ExecutionInput.TargetVersion != "" && result.Binding.PluginVersion != job.ExecutionInput.TargetVersion {
+		return validationError("parsed log batch result does not match leased plugin version")
+	}
+	if result.FirstCursor.SourceIdentityDigest != result.LastCursor.SourceIdentityDigest || result.FirstCursor.StreamGeneration != result.LastCursor.StreamGeneration {
+		return validationError("parsed log batch result crosses source identity or generation boundaries")
+	}
+	return nil
+}
+
 func sameSCUMSchemaProbeBinding(a, b domain.SCUMBindingIdentity) bool {
 	return a.ServerInstanceID == b.ServerInstanceID && a.RunBindingID == b.RunBindingID && a.RunEndpointID == b.RunEndpointID && a.PluginID == b.PluginID && a.PluginVersion == b.PluginVersion && a.AdapterVersion == b.AdapterVersion && a.GameVersion == b.GameVersion && a.DatabaseIdentity == b.DatabaseIdentity
 }